[fix][broker] Pass metadataStoreSessionTimeoutMillis to bookkeeper client configuration to properly configure ZK timeout

[gaozhangmin]

Motivation

I found there is problem related to bk's zktimeout config, the default bk zktimeout of bk is 10s, but
We caculate tickTimeMillis = zktimeout/15.

pulsar/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKSessionWatcher.java

Line 63 in 277835a

this.tickTimeMillis = zk.getSessionTimeout() / 15;

In below future.get, always timeout.

pulsar/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKSessionWatcher.java

Lines 108 to 112 in 277835a

	try {
	zkClientState = future.get(tickTimeMillis, TimeUnit.MILLISECONDS);
	} catch (TimeoutException e) {
	// Consider zk disconnection if zk operation takes more than TICK_TIME
	zkClientState = Watcher.Event.KeeperState.Disconnected;

We should support set bk zktimeut by
bkConf.setZkTimeout((int) conf.getMetadataStoreSessionTimeoutMillis());

Log

2025-08-13T00:14:32,770+0800 [metadata-store-zk-session-watcher-38-1] ^[[33mWARN ^[[m org.apache.pulsar.metadata.impl.ZKSessionWatcher.checkConnectionStatus(ZKSessionWatcher.java:115) - ^[[33mZooKeeper operation timed out, xxxxx:2182/bk-clusters/xxxxx-2000-30000^[[m
2025-08-13T00:14:32,770+0800 [metadata-store-zk-session-watcher-19-1] ^[[33mWARN ^[[m org.apache.pulsar.metadata.impl.ZKSessionWatcher.checkConnectionStatus(ZKSessionWatcher.java:115) - ^[[33mZooKeeper operation timed out, xxxxx:2182/bk-clusters/xxxxxxx-2000-30000^[[m
2025-08-13T00:14:32,771+0800 [metadata-store-zk-session-watcher-19-1] ^[[33mWARN ^[[m org.apache.pulsar.metadata.impl.ZKSessionWatcher.checkState(ZKSessionWatcher.java:161) - ^[[33m[73863839042194267] ZooKeeper client is disconnected. Waiting to reconnect, time remaining = 25.0 seconds^[[m
2025-08-13T00:14:32,771+0800 [metadata-store-zk-session-watcher-38-1] ^[[33mWARN ^[[m org.apache.pulsar.metadata.impl.ZKSessionWatcher.checkState(ZKSessionWatcher.java:161) - ^[[33m[73863839042194268] ZooKeeper client is disconnected. Waiting to reconnect, time remaining = 25.0 seconds^[[m
2025-08-13T00:14:34,772+0800 [metadata-store-zk-session-watcher-38-1] ^[[32mINFO ^[[m org.apache.pulsar.metadata.impl.ZKSessionWatcher.checkState(ZKSessionWatcher.java:173) - ^[[32mZooKeeper client reconnection with server quorum. Current status: ConnectionLost^[[m
2025-08-13T00:14:34,772+0800 [metadata-store-zk-session-watcher-19-1] ^[[32mINFO ^[[m org.apache.pulsar.metadata.impl.ZKSessionWatcher.checkState(ZKSessionWatcher.java:173) - ^[[32mZooKeeper client reconnection with server quorum. Current status: ConnectionLost^[[m

Modifications

bkConf.setZkTimeout((int) conf.getMetadataStoreSessionTimeoutMillis());

Verifying this change

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

If the box was checked, please highlight the changes

• 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

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 74.32%. Comparing base (bbf56f6) to head (43e970d).

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #24624      +/-   ##
============================================
+ Coverage     74.23%   74.32%   +0.09%     
+ Complexity    33179    33177       -2     
============================================
  Files          1881     1881              
  Lines        146850   146851       +1     
  Branches      16866    16866              
============================================
+ Hits         109019   109153     +134     
+ Misses        29191    29027     -164     
- Partials       8640     8671      +31     

Flag	Coverage Δ
inttests	26.78% <100.00%> (?)
systests	23.29% <100.00%> (-0.04%)	⬇️
unittests	73.82% <100.00%> (+0.01%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...che/pulsar/broker/BookKeeperClientFactoryImpl.java	80.18% <100.00%> (+0.18%)	⬆️

... and 90 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[lhotari]

LGTM. The zk timeout setting would only matter when bookkeeperMetadataServiceUri is set.

[lhotari]

@gaozhangmin During a test related PR #24171 I noticed this about the same code location:

pulsar/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKSessionWatcher.java

Lines 108 to 115 in 6e8c349

	try {
	zkClientState = future.get(tickTimeMillis, TimeUnit.MILLISECONDS);
	} catch (TimeoutException e) {
	// Consider zk disconnection if zk operation takes more than TICK_TIME
	zkClientState = Watcher.Event.KeeperState.Disconnected;
	}
	checkState(zkClientState);

This will trigger a ConnectionLost session event. That's why a delay should be kept under 2000 ms.
Currently MockZooKeeper's session timeout is hard coded to 30000ms and the check interval is 1/15 of this, therefore 2000ms.

When there's a test that blocks the ZooKeeper thread for more than 2000ms, it will result in a ConnectionLost event.
This PR makes the session timeout configurable and it's possible to increase the session timeout for tests where it's necessary to use a longer session timeout to avoid ConnectionLost session events.

This applies likewise to production code. This is problematic since the effective timeout becomes 1/15 of the zk session timeout for Pulsar operations.

Together with this change made in #23018 it will add significant load to ZooKeepers:

pulsar/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java

Lines 513 to 519 in 829df71

	// Clear cache after session expired.
	if (event == SessionEvent.SessionReestablished || event == SessionEvent.Reconnected) {
	for (MetadataCacheImpl metadataCache : metadataCaches) {
	metadataCache.invalidateAll();
	}
	invalidateAll();
	}

I'll file a new issue about resolving this problem.

[lhotari]

Issue about ZKSessionWatcher is #24674