You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/06/20 21:00:36 UTC
(pulsar) branch master updated: [fix] functions: Run worker leader-election off the consumer event-listener thread (#26059)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new ced1280a108 [fix] functions: Run worker leader-election off the consumer event-listener thread (#26059)
ced1280a108 is described below
commit ced1280a108709b9fddbb4f8a161fd288c3e2565
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Sat Jun 20 14:00:29 2026 -0700
[fix] functions: Run worker leader-election off the consumer event-listener thread (#26059)
---
.../pulsar/functions/worker/LeaderService.java | 28 +++++++++++++++++++++-
.../pulsar/functions/worker/LeaderServiceTest.java | 8 +++++++
2 files changed, 35 insertions(+), 1 deletion(-)
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/LeaderService.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/LeaderService.java
index cc974d5045c..b2c5a0a5a48 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/LeaderService.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/LeaderService.java
@@ -18,6 +18,11 @@
*/
package org.apache.pulsar.functions.worker;
+import com.google.common.annotations.VisibleForTesting;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.function.Supplier;
import lombok.CustomLog;
import org.apache.pulsar.client.api.Consumer;
@@ -43,6 +48,11 @@ public class LeaderService implements AutoCloseable, ConsumerEventListener {
private final WorkerConfig workerConfig;
private final PulsarClient pulsarClient;
private volatile boolean isLeader = false;
+ // The consumer event listener callbacks (becameActive/becameInactive) run the blocking
+ // leader-election routines on this dedicated single-threaded executor so that the Pulsar client's
+ // shared consumer-listener thread is not blocked. The single thread also preserves event ordering.
+ private final ExecutorService executor =
+ Executors.newSingleThreadExecutor(new DefaultThreadFactory("function-worker-leader"));
static final String COORDINATION_TOPIC_SUBSCRIPTION = "participants";
@@ -90,6 +100,12 @@ public class LeaderService implements AutoCloseable, ConsumerEventListener {
@Override
public void becameActive(Consumer<?> consumer, int partitionId) {
+ // Run the (blocking) become-leader routine on a dedicated executor so the consumer
+ // event-listener thread, which is shared with the Pulsar client, is not blocked.
+ executor.execute(() -> becameActiveInternal(consumer, partitionId));
+ }
+
+ private void becameActiveInternal(Consumer<?> consumer, int partitionId) {
synchronized (this) {
if (isLeader) {
return;
@@ -151,7 +167,11 @@ public class LeaderService implements AutoCloseable, ConsumerEventListener {
}
@Override
- public synchronized void becameInactive(Consumer<?> consumer, int partitionId) {
+ public void becameInactive(Consumer<?> consumer, int partitionId) {
+ executor.execute(() -> becameInactiveInternal(consumer, partitionId));
+ }
+
+ private synchronized void becameInactiveInternal(Consumer<?> consumer, int partitionId) {
if (isLeader) {
log.info().attr("worker", consumerName).log("Worker lost the leadership.");
isLeader = false;
@@ -180,10 +200,16 @@ public class LeaderService implements AutoCloseable, ConsumerEventListener {
return isLeader;
}
+ @VisibleForTesting
+ void joinPendingEventTasks() throws InterruptedException, ExecutionException {
+ executor.submit(() -> { }).get();
+ }
+
@Override
public void close() throws PulsarClientException {
if (consumer != null) {
consumer.close();
}
+ executor.shutdown();
}
}
diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/LeaderServiceTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/LeaderServiceTest.java
index 93e951383f0..f6dea177dc9 100644
--- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/LeaderServiceTest.java
+++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/LeaderServiceTest.java
@@ -138,6 +138,7 @@ public class LeaderServiceTest {
verify(mockClient, times(1)).newConsumer();
listenerHolder.get().becameActive(mockConsumer, 0);
+ leaderService.joinPendingEventTasks();
assertTrue(leaderService.isLeader());
verify(functionMetadataManager, times(1)).getIsInitialized();
@@ -153,6 +154,7 @@ public class LeaderServiceTest {
verify(schedulerManager, times((1))).initialize(any());
listenerHolder.get().becameInactive(mockConsumer, 0);
+ leaderService.joinPendingEventTasks();
assertFalse(leaderService.isLeader());
verify(functionAssignmentTailer, times(1)).startFromMessage(messageId);
@@ -168,6 +170,7 @@ public class LeaderServiceTest {
verify(mockClient, times(1)).newConsumer();
listenerHolder.get().becameActive(mockConsumer, 0);
+ leaderService.joinPendingEventTasks();
assertTrue(leaderService.isLeader());
verify(functionMetadataManager, times(1)).acquireExclusiveWrite(any());
@@ -179,6 +182,7 @@ public class LeaderServiceTest {
verify(schedulerManager, times((1))).initialize(any());
listenerHolder.get().becameInactive(mockConsumer, 0);
+ leaderService.joinPendingEventTasks();
assertFalse(leaderService.isLeader());
verify(functionAssignmentTailer, times(1)).start();
@@ -198,6 +202,7 @@ public class LeaderServiceTest {
when(schedulerManager.acquireExclusiveWrite(any())).thenThrow(new WorkerUtils.NotLeaderAnymore());
listenerHolder.get().becameActive(mockConsumer, 0);
+ leaderService.joinPendingEventTasks();
// should have failed to become leader
assertFalse(leaderService.isLeader());
@@ -214,6 +219,7 @@ public class LeaderServiceTest {
verify(schedulerManager, times((0))).initialize(any());
listenerHolder.get().becameInactive(mockConsumer, 0);
+ leaderService.joinPendingEventTasks();
assertFalse(leaderService.isLeader());
verify(functionAssignmentTailer, times(0)).startFromMessage(messageId);
@@ -234,6 +240,7 @@ public class LeaderServiceTest {
when(functionMetadataManager.acquireExclusiveWrite(any())).thenThrow(new WorkerUtils.NotLeaderAnymore());
listenerHolder.get().becameActive(mockConsumer, 0);
+ leaderService.joinPendingEventTasks();
// should have failed to become leader
assertFalse(leaderService.isLeader());
@@ -250,6 +257,7 @@ public class LeaderServiceTest {
verify(schedulerManager, times((0))).initialize(any());
listenerHolder.get().becameInactive(mockConsumer, 0);
+ leaderService.joinPendingEventTasks();
assertFalse(leaderService.isLeader());
verify(functionAssignmentTailer, times(0)).startFromMessage(messageId);