Merge pull request #8879 from knutwalker/2.6-arrow-authentication-issue

Avoid invalidation of user login state in Arrow server

Author: s1ck

core-utils/src/main/java/org/neo4j/gds/core/utils/ClockService.java

@@ -20,20 +20,31 @@
 package org.neo4j.gds.core.utils;
 
 import java.time.Clock;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
 
 public final class ClockService {
 
     private static final Clock SYSTEM_CLOCK = Clock.systemUTC();
 
-    private static Clock CLOCK = SYSTEM_CLOCK;
+    private static final AtomicReference<Clock> CLOCK = new AtomicReference<>(SYSTEM_CLOCK);
 
     public static void setClock(Clock clock) {
-        CLOCK = clock;
+        CLOCK.set(clock);
     }
 
     public static Clock clock() {
-        return CLOCK;
+        return CLOCK.get();
     }
 
     private ClockService() {}
+
+    public static <T extends Clock> void runWithClock(T clock, Consumer<T> runnable) {
+        Clock previousClock = CLOCK.getAndSet(clock);
+        try {
+            runnable.accept(clock);
+        } finally {
+            CLOCK.set(previousClock);
+        }
+    }
 }

core/src/main/java/org/neo4j/gds/core/concurrency/ExecutorServiceUtil.java

@@ -29,6 +29,7 @@
 import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.FutureTask;
 import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -49,6 +50,10 @@ public static ExecutorService createSingleThreadPool(String threadPrefix) {
         return Executors.newSingleThreadExecutor(NamedThreadFactory.daemon(threadPrefix));
     }
 
+    public static ScheduledExecutorService createSingleThreadScheduler(String threadPrefix) {
+        return Executors.newSingleThreadScheduledExecutor(NamedThreadFactory.daemon(threadPrefix));
+    }
+
     static ExecutorService createThreadPool(int corePoolSize, int maxPoolSize) {
         return createThreadPool(THREAD_NAME_PREFIX, corePoolSize, maxPoolSize);
     }