[fix][metadata]Solve the problem of broker log looping and request failure stagnation after network packet loss recovery

[Radiancebobo]

Motivation

During ZooKeeper session-expired recovery, the broker can close the old ZooKeeper handle and create a new
ZooKeeper client. The old handle may emit KeeperState.Closed as part of this local close sequence.

Before this change, ZKSessionWatcher treated every ZooKeeper KeeperState other than Expired and
Disconnected as a successful reconnection. This means KeeperState.Closed could be translated into
Reconnected and SessionReestablished, even though Closed only means the old ZooKeeper handle was closed. It
does not prove that a new session has connected to the ZooKeeper quorum.

This matters because Reconnected and SessionReestablished are not passive notifications in Pulsar. They trigger
metadata recovery work, including persistent watch recreation, resource lock revalidation, and leader election
revalidation. If these events are published before the new ZooKeeper session is actually healthy, the broker can start
recovery work against an unstable or expired session.

In a severe network-loss scenario, this can become a recovery storm:

1. ZooKeeper session expires.
2. The broker starts the reconnect flow and closes the old ZooKeeper handle.
3. The old handle emits KeeperState.Closed.
4. Closed is incorrectly treated as a successful reconnection.
5. Pulsar publishes Reconnected / SessionReestablished.
6. Watch recreation, lock revalidation, and leader election recovery start too early.
7. These recovery operations fail against the unstable session and can trigger more reconnect/recovery work.

The logs will output in a loop as follows:

PulsarZooKeeperClient - ZooKeeper session 0 is created to pulsar-zookeeper.pulsar:2181.
ZKSessionWatcher - Got ZK session watch event: WatchedEvent state:Closed type:None path:null
PulsarService - Received metadata service session event: Reconnected
ZKMetadataStore - Failed to recreate persistent watch on ZooKeeper: SESSIONEXPIRED
ResourceLockImpl - Failed to revalidate the lock at /namespace/... SessionExpiredException - Retrying ...

The fix is to make the state machine explicit: only KeeperState.SyncConnected should be treated as a successful
reconnection.

Modifications

• Tightened ZKSessionWatcher state handling so only KeeperState.SyncConnected publishes Reconnected and
  SessionReestablished.
• Handled non-recovery states explicitly:
  • Disconnected continues to wait for reconnection.
  • ConnectedReadOnly waits for a writable quorum connection instead of treating the read-only connection as fully
    recovered.
  • AuthFailed marks the metadata session as lost.
  • SaslAuthenticated is logged without publishing recovery events.
  • Closed is logged without changing the metadata session state.
• Added ZKSessionWatcherTest to verify that Closed and other non-SyncConnected states do not publish
  Reconnected or SessionReestablished after SessionLost.
• Kept the existing stale session-id protection in ZKSessionWatcher.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Added unit tests for ZKSessionWatcher state transitions.
• Verified existing ZooKeeper session recovery behavior with ZKSessionTest.

Commands run locally:

./gradlew :pulsar-metadata:test --tests org.apache.pulsar.metadata.impl.ZKSessionWatcherTest --tests org.apache.pulsar.metadata.ZKSessionTest
./gradlew :pulsar-metadata:checkstyleMain :pulsar-metadata:checkstyleTest

Does this pull request potentially affect one of the following parts:

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

[Radiancebobo]

Our team used mirrored images of the old and new code to inject 90% packet loss into each broker pod for half an hour. Without the fix, even after the broker network recovered, the logs would keep looping and the broker would experience prolonged connection interruptions. With the fix applied, the broker traffic recovered quickly after the network was restored. Therefore, I believe this PR is still valuable.

[void-ptr974]

Thanks for the detailed investigation and the clear explanation of the recovery loop. Preventing Closed from the old ZooKeeper handle from being treated as a successful reconnection is a valuable and well-motivated fix.

However, I think the expanded state handling should be completed before merging. The scheduled checkConnectionStatus() probe can currently bypass the new KeeperState switch because every result except CONNECTIONLOSS and SESSIONEXPIRED is converted to SyncConnected.

For example:

1. AuthFailed

   AuthFailed
     → SessionLost
     → the next exists("/") returns AUTHFAILED
     → default maps it to SyncConnected
     → Reconnected + SessionReestablished
   

   This incorrectly reverses the AuthFailed handling on the next scheduled probe.

2. ConnectedReadOnly

   ConnectedReadOnly
     → ConnectionLost
     → exists("/") succeeds because it is a read operation
     → OK is mapped to SyncConnected
     → full recovery is triggered before reaching a writable quorum
   

   Additionally, routing ConnectedReadOnly through handleDisconnected() can publish SessionLost after the timeout even though no Expired event was received. A read-only connection is not fully writable, but it also does not necessarily mean that the session has expired.

I suggest updating the probe so that it preserves the actual ZooKeeper state:

• Map AUTHFAILED explicitly to KeeperState.AuthFailed.
• For successful probes, distinguish CONNECTED from CONNECTEDREADONLY using zk.getState().
• Handle ConnectedReadOnly separately instead of promoting it to SessionLost through the disconnection timeout.
• Only publish SessionReestablished after an actual session loss followed by a writable SyncConnected state.

Please also add deterministic tests for the scheduled probe path, for example:

• AuthFailed followed by an AUTHFAILED probe must not publish Reconnected or SessionReestablished.
• ConnectedReadOnly followed by a successful read probe must remain read-only and must not publish full recovery events.
• ConnectedReadOnly must not become SessionLost without an actual Expired event.
• Closed from the old handle must remain ignored until the replacement client reports a real writable connection.

The current tests only invoke process() directly and close the watcher before the scheduled probe runs, so they do not cover these transitions.

If the intention is to keep this PR narrowly focused, another reasonable option is to handle only the original Closed regression here and address AuthFailed, ConnectedReadOnly, and the probe-state mapping in a separate PR.