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 2024/08/08 08:19:45 UTC
(pulsar) branch master updated: [fix][broker] Fix the bug that elected leader thinks it's a follower (#23138)
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 3560ddb64f4 [fix][broker] Fix the bug that elected leader thinks it's a follower (#23138)
3560ddb64f4 is described below
commit 3560ddb64f44fb2a53d52ef3df0624bb9bda1af6
Author: Heesung Sohn <10...@users.noreply.github.com>
AuthorDate: Thu Aug 8 01:19:38 2024 -0700
[fix][broker] Fix the bug that elected leader thinks it's a follower (#23138)
---
.../coordination/impl/LeaderElectionImpl.java | 15 ++++--
.../apache/pulsar/metadata/LeaderElectionTest.java | 2 +
.../org/apache/pulsar/metadata/ZKSessionTest.java | 55 ++++++++++++++++++++++
3 files changed, 69 insertions(+), 3 deletions(-)
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java
index aa606084173..ab35eb7040c 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/LeaderElectionImpl.java
@@ -134,8 +134,11 @@ class LeaderElectionImpl<T> implements LeaderElection<T> {
// If the value is the same as our proposed value, it means this instance was the leader at some
// point before. The existing value can either be for this same session or for a previous one.
if (res.getStat().isCreatedBySelf()) {
+ log.info("Keeping the existing value {} for {} as it's from the same session stat={}", existingValue,
+ path, res.getStat());
// The value is still valid because it was created in the same session
changeState(LeaderElectionState.Leading);
+ return CompletableFuture.completedFuture(LeaderElectionState.Leading);
} else {
log.info("Conditionally deleting existing equals value {} for {} because it's not created in the "
+ "current session. stat={}", existingValue, path, res.getStat());
@@ -271,7 +274,13 @@ class LeaderElectionImpl<T> implements LeaderElection<T> {
return CompletableFuture.completedFuture(null);
}
- return store.delete(path, version);
+ return store.delete(path, version)
+ .thenAccept(__ -> {
+ synchronized (LeaderElectionImpl.this) {
+ leaderElectionState = LeaderElectionState.NoLeader;
+ }
+ }
+ );
}
@Override
@@ -292,8 +301,8 @@ class LeaderElectionImpl<T> implements LeaderElection<T> {
private void handleSessionNotification(SessionEvent event) {
// Ensure we're only processing one session event at a time.
sequencer.sequential(() -> FutureUtil.composeAsync(() -> {
- if (event == SessionEvent.SessionReestablished) {
- log.info("Revalidating leadership for {}", path);
+ if (event == SessionEvent.Reconnected || event == SessionEvent.SessionReestablished) {
+ log.info("Revalidating leadership for {}, event:{}", path, event);
return elect().thenAccept(leaderState -> {
log.info("Resynced leadership for {} - State: {}", path, leaderState);
}).exceptionally(ex -> {
diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LeaderElectionTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LeaderElectionTest.java
index 6b4f74a30b5..4b48f3c20b0 100644
--- a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LeaderElectionTest.java
+++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LeaderElectionTest.java
@@ -69,6 +69,8 @@ public class LeaderElectionTest extends BaseMetadataStoreTest {
leaderElection.close();
+ assertEquals(leaderElection.getState(), LeaderElectionState.NoLeader);
+
assertEquals(cache.get("/my/leader-election").join(), Optional.empty());
}
diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java
index 36cb0f132ba..02d65fd21ed 100644
--- a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java
+++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java
@@ -27,6 +27,7 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import lombok.Cleanup;
+import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
import org.apache.pulsar.metadata.api.coordination.CoordinationService;
import org.apache.pulsar.metadata.api.coordination.LeaderElection;
@@ -180,4 +181,58 @@ public class ZKSessionTest extends BaseMetadataStoreTest {
.untilAsserted(()-> assertEquals(le1.getState(),LeaderElectionState.Leading));
assertTrue(store.get(path).join().isPresent());
}
+
+
+ @Test
+ public void testElectAfterReconnected() throws Exception {
+ // --- init
+ @Cleanup
+ MetadataStoreExtended store = MetadataStoreExtended.create(zks.getConnectionString(),
+ MetadataStoreConfig.builder()
+ .sessionTimeoutMillis(2_000)
+ .build());
+
+
+ BlockingQueue<SessionEvent> sessionEvents = new LinkedBlockingQueue<>();
+ store.registerSessionListener(sessionEvents::add);
+ BlockingQueue<LeaderElectionState> leaderElectionEvents = new LinkedBlockingQueue<>();
+ String path = newKey();
+
+ @Cleanup
+ CoordinationService coordinationService = new CoordinationServiceImpl(store);
+ @Cleanup
+ LeaderElection<String> le1 = coordinationService.getLeaderElection(String.class, path,
+ leaderElectionEvents::add);
+
+ // --- test manual elect
+ String proposed = "value-1";
+ le1.elect(proposed).join();
+ assertEquals(le1.getState(), LeaderElectionState.Leading);
+ LeaderElectionState les = leaderElectionEvents.poll(5, TimeUnit.SECONDS);
+ assertEquals(les, LeaderElectionState.Leading);
+
+
+ // simulate no leader state
+ FieldUtils.writeDeclaredField(le1, "leaderElectionState", LeaderElectionState.NoLeader, true);
+
+ // reconnect
+ zks.stop();
+
+ SessionEvent e = sessionEvents.poll(5, TimeUnit.SECONDS);
+ assertEquals(e, SessionEvent.ConnectionLost);
+
+ zks.start();
+
+
+ // --- test le1 can be leader
+ e = sessionEvents.poll(10, TimeUnit.SECONDS);
+ assertEquals(e, SessionEvent.Reconnected);
+ Awaitility.await().atMost(Duration.ofSeconds(15))
+ .untilAsserted(()-> {
+ assertEquals(le1.getState(),LeaderElectionState.Leading);
+ }); // reacquire leadership
+
+
+ assertTrue(store.get(path).join().isPresent());
+ }
}