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/13 06:04:26 UTC
(pulsar) 07/08: [fix][meta] Don't lose SessionReestablished when a listener invalidates the new session (#26861)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c886f94f61dcc6973ae196e8d028a0d2471409e1
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Mon Oct 12 12:18:02 2026 -0700
[fix][meta] Don't lose SessionReestablished when a listener invalidates the new session (#26861)
(cherry picked from commit cc346d2c0eb7d4fd6a1a4d4c8f5a9c45f8781660)
---
.../pulsar/metadata/impl/ZKMetadataStore.java | 10 ++-
.../pulsar/metadata/impl/ZKSessionWatcher.java | 42 ++++++++++-
.../pulsar/metadata/impl/ZKSessionWatcherTest.java | 88 ++++++++++++++++++++++
3 files changed, 134 insertions(+), 6 deletions(-)
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java
index f6c22a291438..1ba8eb4e3bb5 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java
@@ -171,11 +171,17 @@ public class ZKMetadataStore extends AbstractBatchedMetadataStore
super.receivedSessionEvent(event);
} else {
log.error("Failed to recreate persistent watch on ZooKeeper: {}", Code.get(rc));
+ // Only another session may reestablish the session on the re-connectable client, which
+ // replaces it below
if (sessionWatcher != null) {
- sessionWatcher.setSessionInvalid();
+ if (zkc instanceof PulsarZooKeeperClient) {
+ sessionWatcher.setSessionReplaced();
+ } else {
+ sessionWatcher.setSessionInvalid();
+ }
}
// On the re-connectable client, mark the session as expired to trigger a new reconnect,
- // and we will have the chance to set the watch again.
+ // and we will have the chance to set the watch again on the new session.
if (zkc instanceof PulsarZooKeeperClient) {
((PulsarZooKeeperClient) zkc).process(
new WatchedEvent(Watcher.Event.EventType.None,
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKSessionWatcher.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKSessionWatcher.java
index cc26231a7452..7b6739799b2d 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKSessionWatcher.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKSessionWatcher.java
@@ -57,6 +57,11 @@ public class ZKSessionWatcher implements AutoCloseable, Watcher {
private long disconnectedAt = 0;
+ // The id of a session that a listener invalidated and that PulsarZooKeeperClient is replacing, or 0. Events of
+ // that session, such as a connection check that still succeeds before the session is closed, must not
+ // reestablish it, or the replacing session's events would find it reestablished already and send nothing.
+ private long replacedSessionId = 0;
+
public ZKSessionWatcher(ZooKeeper zk, Consumer<SessionEvent> sessionListener) {
this.zk = zk;
this.monitorTimeoutMillis = zk.getSessionTimeout() * 5 / 6;
@@ -131,6 +136,15 @@ public class ZKSessionWatcher implements AutoCloseable, Watcher {
currentStatus = SessionEvent.SessionLost;
}
+ /**
+ * Mark the current session invalid while it's being replaced with a new session: only the new session's
+ * connection reestablishes the session.
+ */
+ synchronized void setSessionReplaced() {
+ setSessionInvalid();
+ replacedSessionId = zk.getSessionId();
+ }
+
// PulsarZooKeeperClient publishes the new ZooKeeper instance before forwarding the corresponding session event to
// watcherManager, so zk.set(newZk) happens-before this watcher observes the new-session event. Keep the session-id
// check and state transition in the same synchronized section to prevent stale async probes from racing with that
@@ -180,17 +194,37 @@ public class ZKSessionWatcher implements AutoCloseable, Watcher {
}
break;
+ case Closed:
+ // A ZooKeeper handle was closed: the previous handle when PulsarZooKeeperClient replaces an expired
+ // session with a new one, or the client when the metadata store closes. It says nothing about the
+ // connection of the current handle, whose own events report the new session.
+ break;
+
default:
if (currentStatus != SessionEvent.SessionReestablished) {
+ if (replacedSessionId != 0 && zk.getSessionId() == replacedSessionId) {
+ log.info("Ignoring the connection of a ZooKeeper session that is being replaced, sessionId={}",
+ replacedSessionId);
+ break;
+ }
+ replacedSessionId = 0;
// since it reconnected to zoo keeper, we reset the disconnected time
log.info("ZooKeeper client reconnection with server quorum. Current status: {}", currentStatus);
disconnectedAt = 0;
- sessionListener.accept(SessionEvent.Reconnected);
- if (currentStatus == SessionEvent.SessionLost) {
- sessionListener.accept(SessionEvent.SessionReestablished);
- }
+ SessionEvent previousStatus = currentStatus;
+ // Update the status before notifying the listener: a listener that can't use the new session, such
+ // as ZKMetadataStore when it fails to recreate its persistent watch, calls setSessionReplaced() or
+ // setSessionInvalid() while it's notified, and the status it sets must not be overwritten, or the next
+ // session would not send SessionReestablished again.
currentStatus = SessionEvent.SessionReestablished;
+ try {
+ sessionListener.accept(SessionEvent.Reconnected);
+ } finally {
+ if (previousStatus == SessionEvent.SessionLost) {
+ sessionListener.accept(SessionEvent.SessionReestablished);
+ }
+ }
}
break;
}
diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/ZKSessionWatcherTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/ZKSessionWatcherTest.java
new file mode 100644
index 000000000000..0fd200cf788a
--- /dev/null
+++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/ZKSessionWatcherTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.impl;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import lombok.Cleanup;
+import org.apache.pulsar.metadata.api.extended.SessionEvent;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooKeeper;
+import org.testng.annotations.Test;
+
+public class ZKSessionWatcherTest {
+
+ /**
+ * When ZKMetadataStore can't use a reestablished session, such as when it fails to recreate its persistent watch,
+ * it marks the session replaced while it's notified and makes PulsarZooKeeperClient replace the session. The new
+ * session must send SessionReestablished again, so that the listeners, such as the lock managers that revalidate
+ * their locks, get it, and the replaced session's own events must not reestablish it first.
+ */
+ @Test(timeOut = 30_000)
+ public void sessionReplacedWhileNotifiedIsReestablishedWithNextSession() throws Exception {
+ ZooKeeper zk = mock(ZooKeeper.class);
+ // a long session timeout, so that the periodic connection check doesn't run during the test
+ when(zk.getSessionTimeout()).thenReturn(600_000);
+ AtomicLong sessionId = new AtomicLong(1);
+ when(zk.getSessionId()).thenAnswer(invocation -> sessionId.get());
+ List<SessionEvent> events = new CopyOnWriteArrayList<>();
+ AtomicReference<ZKSessionWatcher> watcher = new AtomicReference<>();
+ AtomicBoolean replaceOnce = new AtomicBoolean(true);
+ @Cleanup
+ ZKSessionWatcher sessionWatcher = new ZKSessionWatcher(zk, event -> {
+ events.add(event);
+ if (event == SessionEvent.SessionReestablished && replaceOnce.getAndSet(false)) {
+ watcher.get().setSessionReplaced();
+ }
+ });
+ watcher.set(sessionWatcher);
+
+ // the session expired, and PulsarZooKeeperClient replaced it with session 1, closing the previous handle
+ sessionWatcher.process(sessionEvent(Watcher.Event.KeeperState.Expired));
+ sessionWatcher.process(sessionEvent(Watcher.Event.KeeperState.Closed));
+ assertThat(events).containsExactly(SessionEvent.SessionLost);
+ sessionWatcher.process(sessionEvent(Watcher.Event.KeeperState.SyncConnected));
+ assertThat(events).containsExactly(SessionEvent.SessionLost, SessionEvent.Reconnected,
+ SessionEvent.SessionReestablished);
+
+ // The listener marked session 1 replaced. A connection check of session 1 that still succeeds, and the
+ // Closed event of its handle, don't reestablish it.
+ sessionWatcher.process(sessionEvent(Watcher.Event.KeeperState.SyncConnected));
+ sessionWatcher.process(sessionEvent(Watcher.Event.KeeperState.Closed));
+ assertThat(events).containsExactly(SessionEvent.SessionLost, SessionEvent.Reconnected,
+ SessionEvent.SessionReestablished);
+
+ // session 2 connects
+ sessionId.set(2);
+ sessionWatcher.process(sessionEvent(Watcher.Event.KeeperState.SyncConnected));
+ assertThat(events).containsExactly(SessionEvent.SessionLost, SessionEvent.Reconnected,
+ SessionEvent.SessionReestablished, SessionEvent.Reconnected, SessionEvent.SessionReestablished);
+ }
+
+ private static WatchedEvent sessionEvent(Watcher.Event.KeeperState state) {
+ return new WatchedEvent(Watcher.Event.EventType.None, state, null);
+ }
+}