[GitHub] [pulsar] jerrypeng commented on a change in pull request #7237: Fix leader/scheduler assignment processing lag problem

[GitBox <gi...@apache.org>]

jerrypeng commented on a change in pull request #7237:
URL: https://github.com/apache/pulsar/pull/7237#discussion_r439266467



##########
File path: pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionAssignmentTailer.java
##########
@@ -70,33 +78,64 @@ public FunctionAssignmentTailer(
                             log.warn("Encountered error when assignment tailer is not running", th);
                         }
                     }
-
                 }
             }
+            log.info("tailer thread exiting...");
+            hasExited.complete(null);
         });
         this.tailerThread.setName("assignment-tailer-thread");
     }
 
-    public void start() {
-        isRunning = true;
-        tailerThread.start();
+    public CompletableFuture<Void> triggerReadToTheEndAndExit() {
+        exitOnEndOfTopic = true;
+        return this.hasExited;
+    }
+
+    public synchronized void start() throws PulsarClientException {
+        if (!isRunning) {
+            isRunning = true;
+            if (reader == null) {
+                reader = createReader();
+            }
+            tailerThread.start();
+        }
     }
+    
 
     @Override
-    public void close() {
-        log.info("Stopping function assignment tailer");
+    public synchronized void close() {
+        log.info("Closing function assignment tailer");
         try {
             isRunning = false;
-            if (tailerThread != null && tailerThread.isAlive()) {
-                tailerThread.interrupt();
-            }
+
+            if (tailerThread != null) {
+                while (true) {
+                    tailerThread.interrupt();
+
+                    try {
+                        tailerThread.join(5000, 0);
+                    } catch (InterruptedException e) {
+                        log.warn("Waiting for assignment tailer thread to stop is interrupted", e);
+                    }
+
+                    if (tailerThread.isAlive()) {
+                        log.warn("Assignment tailer thread is still alive.  Will attempt to interrupt again.");
+                    } else {
+                        break;
+                    }
+                }
+            }            
             if (reader != null) {
                 reader.close();
+                reader = null;
             }
+
+            hasExited = new CompletableFuture<>();

Review comment:
       it is re-initializing the variable, so if we can "start" again, the completeable future is not already completed




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org