From b8fff3cff06606588ef5c29e8e951ca6f6c61e8d Mon Sep 17 00:00:00 2001 From: hechengbo Date: Tue, 28 Jul 2026 10:45:44 +0800 Subject: [PATCH] [fix][metadata]ZKSessionWatcher to only trigger reconnection on SyncConnected state --- .../metadata/impl/ZKSessionWatcher.java | 92 +++++++++++++------ .../metadata/impl/ZKSessionWatcherTest.java | 89 ++++++++++++++++++ 2 files changed, 154 insertions(+), 27 deletions(-) create mode 100644 pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/ZKSessionWatcherTest.java 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 08d87eb94b274..07c3508c60624 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 @@ -160,43 +160,81 @@ private synchronized void checkState(Watcher.Event.KeeperState zkClientState) { break; case Disconnected: - if (disconnectedAt == 0) { - // this is the first disconnect event, we should monitor the time out from now, so we record the - // time of disconnect - disconnectedAt = System.nanoTime(); - } + case ConnectedReadOnly: + handleDisconnected(zkClientState); + break; + + case SyncConnected: + handleSyncConnected(); + break; - long timeRemainingMillis = monitorTimeoutMillis - - TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - disconnectedAt); - if (timeRemainingMillis <= 0 && currentStatus != SessionEvent.SessionLost) { - log.error("ZooKeeper session reconnection timeout. Notifying session is lost."); + case AuthFailed: + if (currentStatus != SessionEvent.SessionLost) { + log.error("ZooKeeper client authentication failed. Notifying session is lost."); currentStatus = SessionEvent.SessionLost; sessionListener.accept(currentStatus); - } else if (currentStatus != SessionEvent.SessionLost) { + } + break; + + case SaslAuthenticated: + log.info().attr("currentStatus", currentStatus) + .log("ZooKeeper client SASL authentication completed"); + break; + + case Closed: + log.info().attr("currentStatus", currentStatus).log("ZooKeeper client is closed"); + break; + + default: + log.warn().attr("zkClientState", zkClientState).attr("currentStatus", currentStatus) + .log("Ignoring ZooKeeper client state that does not indicate a reconnection"); + break; + } + } + + private void handleDisconnected(Watcher.Event.KeeperState zkClientState) { + if (disconnectedAt == 0) { + // this is the first disconnect event, we should monitor the time out from now, so we record the + // time of disconnect + disconnectedAt = System.nanoTime(); + } + + long timeRemainingMillis = monitorTimeoutMillis + - TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - disconnectedAt); + if (timeRemainingMillis <= 0 && currentStatus != SessionEvent.SessionLost) { + log.error("ZooKeeper session reconnection timeout. Notifying session is lost."); + currentStatus = SessionEvent.SessionLost; + sessionListener.accept(currentStatus); + } else if (currentStatus != SessionEvent.SessionLost) { + if (zkClientState == Watcher.Event.KeeperState.ConnectedReadOnly) { + log.warn() + .attr("sessionId", zk.getSessionId()) + .attr("timeRemainingSeconds", timeRemainingMillis / 1000.0) + .log("ZooKeeper client is connected to a read-only server. Waiting for read-write connection"); + } else { log.warn() .attr("sessionId", zk.getSessionId()) .attr("timeRemainingSeconds", timeRemainingMillis / 1000.0) .log("ZooKeeper client is disconnected. Waiting to reconnect"); - if (currentStatus == SessionEvent.SessionReestablished) { - currentStatus = SessionEvent.ConnectionLost; - sessionListener.accept(currentStatus); - } } - break; + if (currentStatus == SessionEvent.SessionReestablished) { + currentStatus = SessionEvent.ConnectionLost; + sessionListener.accept(currentStatus); + } + } + } - default: - if (currentStatus != SessionEvent.SessionReestablished) { - // since it reconnected to zoo keeper, we reset the disconnected time - log.info().attr("currentStatus", currentStatus).log("ZooKeeper client reconnection with server quorum"); - disconnectedAt = 0; - - sessionListener.accept(SessionEvent.Reconnected); - if (currentStatus == SessionEvent.SessionLost) { - sessionListener.accept(SessionEvent.SessionReestablished); - } - currentStatus = SessionEvent.SessionReestablished; + private void handleSyncConnected() { + if (currentStatus != SessionEvent.SessionReestablished) { + // since it reconnected to zoo keeper, we reset the disconnected time + log.info().attr("currentStatus", currentStatus).log("ZooKeeper client reconnection with server quorum"); + disconnectedAt = 0; + + sessionListener.accept(SessionEvent.Reconnected); + if (currentStatus == SessionEvent.SessionLost) { + sessionListener.accept(SessionEvent.SessionReestablished); } - break; + currentStatus = SessionEvent.SessionReestablished; } } } 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 0000000000000..8b8eca1ff98eb --- /dev/null +++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/ZKSessionWatcherTest.java @@ -0,0 +1,89 @@ +/* + * 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.mockito.Mockito.mock; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import org.apache.pulsar.metadata.api.extended.SessionEvent; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher.Event.EventType; +import org.apache.zookeeper.Watcher.Event.KeeperState; +import org.apache.zookeeper.ZooKeeper; +import org.testng.annotations.Test; + +@Test +public class ZKSessionWatcherTest { + + @Test + public void testClosedEventShouldNotBeTreatedAsReconnectedAfterSessionLost() throws Exception { + List events = new CopyOnWriteArrayList<>(); + try (ZKSessionWatcher watcher = newSessionWatcher(events)) { + watcher.setSessionInvalid(); + watcher.process(new WatchedEvent(EventType.None, KeeperState.Closed, null)); + + assertTrue(events.isEmpty(), + "Closed is a terminal state for the old ZooKeeper handle and must not be treated as " + + "Reconnected or SessionReestablished, but received " + events); + } + } + + @Test + public void testOnlySyncConnectedShouldBeTreatedAsReconnectedAfterSessionLost() throws Exception { + List events = new CopyOnWriteArrayList<>(); + try (ZKSessionWatcher watcher = newSessionWatcher(events)) { + watcher.setSessionInvalid(); + watcher.process(new WatchedEvent(EventType.None, KeeperState.SyncConnected, null)); + + assertEquals(events, Arrays.asList(SessionEvent.Reconnected, SessionEvent.SessionReestablished)); + } + } + + @Test + public void testNonSyncConnectedEventsShouldNotBeTreatedAsReconnectedAfterSessionLost() throws Exception { + for (KeeperState keeperState : Arrays.asList( + KeeperState.Disconnected, + KeeperState.AuthFailed, + KeeperState.ConnectedReadOnly, + KeeperState.SaslAuthenticated, + KeeperState.Closed)) { + List events = new CopyOnWriteArrayList<>(); + try (ZKSessionWatcher watcher = newSessionWatcher(events)) { + watcher.setSessionInvalid(); + watcher.process(new WatchedEvent(EventType.None, keeperState, null)); + + assertTrue(events.stream().noneMatch(event -> event == SessionEvent.Reconnected + || event == SessionEvent.SessionReestablished), + keeperState + " must not be treated as a ZooKeeper reconnection event, but received " + + events); + } + } + } + + private static ZKSessionWatcher newSessionWatcher(List events) { + ZooKeeper zk = mock(ZooKeeper.class); + when(zk.getSessionTimeout()).thenReturn(30_000); + when(zk.getSessionId()).thenReturn(0x1234L); + return new ZKSessionWatcher(zk, events::add); + } +}