adding SchedulerManager tests (#276)

apache · Mar 4, 2018 · 03df440 · 03df440
1 parent 46bc558
commit 03df440
Show file tree

Hide file tree

Showing 2 changed files with 708 additions and 3 deletions.
diff --git a/...r-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java b/...r-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
@@ -41,6 +41,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -87,11 +88,11 @@ public SchedulerManager(WorkerConfig workerConfig, PulsarClient pulsarClient) {
 
         this.executorService =
                 new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
-                        new LinkedBlockingQueue<Runnable>());
+                        new LinkedBlockingQueue<>());
     }
 
-    public void schedule() {
-        executorService.submit(() -> {
+    public Future<?> schedule() {
+        return executorService.submit(() -> {
             synchronized (SchedulerManager.this) {
                 boolean isLeader = membershipManager.isLeader();
                 if (isLeader) {
@@ -164,6 +165,7 @@ private void invokeScheduler() {
 
         // wait for assignment update to go throw the pipeline
         while (this.functionRuntimeManager.getCurrentAssignmentVersion() < assignmentVersion) {
+            log.info("Waiting for assignments to propagate...");
             try {
                 Thread.sleep(500);
             } catch (InterruptedException e) {