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/10/20 13:20:44 UTC
(pulsar) 06/19: [fix][meta] Fix leader election close deleting a leader node owned by another session (#26828)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2-cherry-picks-2026-10-20
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 238afbd26dc7c86bc66465fcc7435e9fef78b4a4
Author: Dream95 <86...@qq.com>
AuthorDate: Sat Oct 17 21:48:19 2026 +0800
[fix][meta] Fix leader election close deleting a leader node owned by another session (#26828)
Signed-off-by: Dream95 <zh...@163.com>
(cherry picked from commit 24c442abb5a5f85a2ba27368e60d2d0c04929486)
---
.../coordination/impl/LeaderElectionImpl.java | 29 ++++-
.../LeaderElectionStaleLeaderCloseTest.java | 144 +++++++++++++++++++++
2 files changed, 169 insertions(+), 4 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 b2e6341b5b60..661987944d47 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
@@ -36,6 +36,7 @@ import org.apache.pulsar.metadata.api.MetadataSerde;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.MetadataStoreException.AlreadyClosedException;
import org.apache.pulsar.metadata.api.MetadataStoreException.BadVersionException;
+import org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException;
import org.apache.pulsar.metadata.api.Notification;
import org.apache.pulsar.metadata.api.NotificationType;
import org.apache.pulsar.metadata.api.coordination.LeaderElection;
@@ -55,7 +56,6 @@ class LeaderElectionImpl<T> implements LeaderElection<T> {
private final Consumer<LeaderElectionState> stateChangesListener;
private LeaderElectionState leaderElectionState;
- private Optional<Long> version = Optional.empty();
private Optional<T> proposedValue;
// The leader value as known by the election cycle (the leader can only change through an
@@ -313,12 +313,33 @@ class LeaderElectionImpl<T> implements LeaderElection<T> {
return CompletableFuture.completedFuture(null);
}
- return store.delete(path, version)
+ // Only delete the leader node if it is still ours: if our session expired, the node we
+ // created is already gone and the node at the path, if any, belongs to the participant
+ // that won the re-election. An unconditional delete would remove that node and make the
+ // new leader lose leadership for no reason.
+ return store.get(path)
+ .thenCompose(optRes -> {
+ if (optRes.isEmpty() || !optRes.get().getStat().isCreatedBySelf()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ return store.delete(path, Optional.of(optRes.get().getStat().getVersion()))
+ .exceptionally(ex -> {
+ Throwable t = FutureUtil.unwrapCompletionException(ex);
+ if (t instanceof NotFoundException || t instanceof BadVersionException) {
+ // A version check detects an in-place update. It does not close the
+ // delete-and-recreate race for stores that reset versions on recreation
+ // (ZooKeeper, LocalMemory, and RocksDB); stores such as Oxia assign a
+ // new version to the recreated record.
+ return null;
+ }
+ throw FutureUtil.wrapToCompletionException(ex);
+ });
+ })
.thenAccept(__ -> {
synchronized (LeaderElectionImpl.this) {
leaderElectionState = LeaderElectionState.NoLeader;
- // The deleted leader node was ours and a closed instance no longer
- // observes elections; don't keep reporting ourselves as leader.
+ // Whether or not we deleted the node, a closed instance no longer observes
+ // elections; don't keep reporting ourselves as leader.
currentLeaderFuture = CompletableFuture.completedFuture(Optional.empty());
}
}
diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LeaderElectionStaleLeaderCloseTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LeaderElectionStaleLeaderCloseTest.java
new file mode 100644
index 000000000000..8b783381777a
--- /dev/null
+++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/LeaderElectionStaleLeaderCloseTest.java
@@ -0,0 +1,144 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.metadata;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataStoreConfig;
+import org.apache.pulsar.metadata.api.coordination.CoordinationService;
+import org.apache.pulsar.metadata.api.coordination.LeaderElection;
+import org.apache.pulsar.metadata.api.coordination.LeaderElectionState;
+import org.apache.pulsar.metadata.coordination.impl.CoordinationServiceImpl;
+import org.apache.pulsar.metadata.impl.ZKMetadataStore;
+import org.apache.zookeeper.MockZooKeeper;
+import org.apache.zookeeper.MockZooKeeperSession;
+import org.testng.annotations.Test;
+
+/**
+ * Tests for closing a leader election instance whose view of the leader node is stale: a metadata
+ * store session expiry removes the ephemeral leader node server-side, and another participant can
+ * create its own leader node before the former leader observes the Deleted notification. Closing
+ * such a stale leader must never delete the node created by the other session.
+ */
+public class LeaderElectionStaleLeaderCloseTest {
+
+ private static final String PATH = "/test/stale-leader-close/leader";
+
+ @Test(timeOut = 30000)
+ public void closeOfActualLeaderStillDeletesItsOwnNode() throws Exception {
+ @Cleanup
+ MockZooKeeper mockZooKeeper = MockZooKeeper.newInstance();
+ @Cleanup
+ MockZooKeeperSession session1 = MockZooKeeperSession.newInstance(mockZooKeeper, false);
+ @Cleanup
+ ZKMetadataStore store1 = new ZKMetadataStore(session1, MetadataStoreConfig.builder().build(), true);
+
+ @Cleanup
+ CoordinationService cs1 = new CoordinationServiceImpl(store1);
+ @Cleanup
+ LeaderElection<String> le1 = cs1.getLeaderElection(String.class, PATH, __ -> { });
+
+ assertEquals(le1.elect("test-1").join(), LeaderElectionState.Leading);
+ assertTrue(store1.get(PATH).join().orElseThrow().getStat().isCreatedBySelf());
+
+ le1.close();
+
+ assertEquals(store1.get(PATH).join(), Optional.empty());
+ assertEquals(le1.getState(), LeaderElectionState.NoLeader);
+ }
+
+ @Test(timeOut = 30000)
+ public void closeOfStaleLeaderAfterFollowerTakeoverDoesNotDeleteTheNewLeadersNode() throws Exception {
+ @Cleanup
+ MockZooKeeper mockZooKeeper = MockZooKeeper.newInstance();
+ @Cleanup
+ MockZooKeeperSession session1 = MockZooKeeperSession.newInstance(mockZooKeeper, false);
+ @Cleanup
+ MockZooKeeperSession session2 = MockZooKeeperSession.newInstance(mockZooKeeper, false);
+ @Cleanup
+ ZKMetadataStore store1 = new ZKMetadataStore(session1, MetadataStoreConfig.builder().build(), true);
+ @Cleanup
+ ZKMetadataStore store2 = new ZKMetadataStore(session2, MetadataStoreConfig.builder().build(), true);
+
+ @Cleanup
+ CoordinationService cs1 = new CoordinationServiceImpl(store1);
+ @Cleanup
+ CoordinationService cs2 = new CoordinationServiceImpl(store2);
+
+ List<LeaderElectionState> le1States = new CopyOnWriteArrayList<>();
+ List<LeaderElectionState> le2States = new CopyOnWriteArrayList<>();
+ CountDownLatch le2Leading = new CountDownLatch(1);
+
+ @Cleanup
+ LeaderElection<String> le1 = cs1.getLeaderElection(String.class, PATH, le1States::add);
+ @Cleanup
+ LeaderElection<String> le2 = cs2.getLeaderElection(String.class, PATH, state -> {
+ le2States.add(state);
+ if (state == LeaderElectionState.Leading) {
+ le2Leading.countDown();
+ }
+ });
+
+ assertEquals(le1.elect("test-1").join(), LeaderElectionState.Leading);
+ assertEquals(le2.elect("test-2").join(), LeaderElectionState.Following);
+
+ // session1 expires: the server drops the dead session's watches, removes its ephemeral node
+ // and delivers the NodeDeleted watch to the remaining live sessions. Only le2's session is
+ // still alive, so only le2 learns about the deletion.
+ mockZooKeeper.deleteWatchers(session1.getSessionId());
+ mockZooKeeper.delete(PATH, -1);
+
+ // le2 takes over through its own machinery: Deleted notification -> re-election -> creating
+ // its own ephemeral leader node.
+ assertTrue(le2Leading.await(20, TimeUnit.SECONDS));
+ assertEquals(le2.getState(), LeaderElectionState.Leading);
+ assertEquals(le2.getLeaderValueIfPresent(), Optional.of("test-2"));
+
+ // le1 saw nothing and still believes it is the leader.
+ assertEquals(le1States, List.of(LeaderElectionState.Leading));
+ assertEquals(le1.getState(), LeaderElectionState.Leading);
+ GetResult foreignNode = store1.get(PATH).join().orElse(null);
+ assertNotNull(foreignNode);
+ assertFalse(foreignNode.getStat().isCreatedBySelf(),
+ "precondition: from le1's session the node must be foreign");
+
+ le1.close();
+
+ GetResult survivor = store1.get(PATH).join().orElse(null);
+ assertNotNull(survivor, "the stale leader's close deleted the node created by the new leader");
+ assertTrue(store2.get(PATH).join().orElseThrow().getStat().isCreatedBySelf(),
+ "the surviving node must be the one le2's session created");
+ assertEquals(le1.getState(), LeaderElectionState.NoLeader);
+ // le2 may recreate the node after receiving the Deleted notification, so the survivor
+ // assertion above can still pass if le1.close() deleted it. The state sequence below
+ // catches the extra re-election and Leading notification caused by that deletion.
+ assertEquals(le2States, List.of(LeaderElectionState.Following, LeaderElectionState.Leading));
+ assertEquals(le2.getState(), LeaderElectionState.Leading);
+ }
+
+}