Add timeout to hasMessageAvailable to leader election process in Puls…

…ar Functions (apache#8687) Co-authored-by: Jerry Peng <[email protected]>
drianpauL · Nov 27, 2020 · 9443ba2 · 9443ba2
1 parent ea5882a
commit 9443ba2
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/...ons/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionAssignmentTailer.java b/...ons/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionAssignmentTailer.java
@@ -145,7 +145,7 @@ private Thread getTailerThread() {
                 try {
                     Message<byte[]> msg = reader.readNext(1, TimeUnit.SECONDS);
                     if (msg == null) {
-                        if (exitOnEndOfTopic && !reader.hasMessageAvailable()) {
+                        if (exitOnEndOfTopic && !reader.hasMessageAvailableAsync().get(10, TimeUnit.SECONDS)) {
                             break;
                         }
                     } else {

diff --git a/.../worker/src/main/java/org/apache/pulsar/functions/worker/FunctionMetaDataTopicTailer.java b/.../worker/src/main/java/org/apache/pulsar/functions/worker/FunctionMetaDataTopicTailer.java
@@ -67,7 +67,7 @@ public void run() {
             try {
                 Message<byte[]> msg = reader.readNext(1, TimeUnit.SECONDS);
                 if (msg == null) {
-                    if (exitOnEndOfTopic && !reader.hasMessageAvailable()) {
+                    if (exitOnEndOfTopic && !reader.hasMessageAvailableAsync().get(10, TimeUnit.SECONDS)) {
                         break;
                     }
                 } else {