Parallelize per-instance listener registration during controller leadership [CICP-34606]

[kabragaurav]

Summary

Parallelize per-instance ZK watch registration during controller leadership acquisition. The bottleneck is in checkLiveInstancesObservation() which sequentially registers currentState, taskCurrentState, message, and customizedStateRoot listeners for every live instance during the INIT callback. For large clusters like venice-5 (300 instances), this means ~1200 sequential ZK roundtrips (~200ms each) = ~240 seconds, causing SLA-0 MissingTopState alerts on KSAP.

Three-part fix:

1. CallbackHandler: Add deferred-init constructor so handler creation is fast and init() can be called separately.

2. ZKHelixManager.addListener(): Move init() outside synchronized(this). This reduces lock hold time from ~200ms (ZK roundtrip) to ~1ms (handler creation), allowing parallel threads to overlap their ZK subscriptions. This is a global change affecting all addXxxListener callers (participants, spectators, controllers). Between _handlers.add() and init() completing, the handler is visible but not yet initialized. Safe because reset() on an uninitialized handler is a no-op, and init() has its own synchronized(this) guard.

3. GenericHelixController.checkLiveInstancesObservation(): During INIT, collect pending instances instead of registering listeners inline. ZKHelixManager.handleNewSession(): After handleNewSessionAsController() returns and invoke() releases synchronized(_manager), register per-instance listeners in parallel using a 10-thread pool. initHandlers() skips already-initialized handlers to avoid spurious WARN logs.

JIRA: https://linkedin.atlassian.net/browse/CICP-34606
RCA doc: https://docs.google.com/document/d/1la09_BYyE77cTL9ruxgGW3u9llWMMAprYQ31654SUC0/edit
Testing doc: https://docs.google.com/document/d/14LoOytLq4-YwwcxjRPCoxvWI6520kQj3JcviL5SYX8k/edit?tab=t.0#heading=h.jt9jh4tz6tl9

Benchmark Results

TestRegisterPendingListenersBenchmark.java

• calls the real registerPendingInstanceListeners() method via doCallRealMethod(). 100 instances, 200ms simulated ZK latency per addXxxListener call.

Run: mvn test -pl helix-core -Dtest=TestRegisterPendingListenersBenchmark -DfailIfNoTests=false -am

	Time	Speedup
Sequential (old: for-loop in checkLiveInstancesObservation)	40,692 ms	1.0x
Parallel (new: registerPendingInstanceListeners, 20 threads)	2,048 ms	19.9x

Latency simulated with Thread.sleep(200) per addXxxListener call to mimic production ZK roundtrips (~200ms avg from acquired leadership took: logs). Local in-memory ZK is too fast (~0ms per call) to show the difference.

Expected production improvement: For venice-5 (300 instances, ~1200 listener registrations), leadership acquisition drops from ~240s to ~24s.

Note: the shipped pool size is 10 (constant INIT_HANDLERS_PARALLELISM). The mock benchmark table above was run with 20 threads; with 10 the parallel time is roughly 2x the benchmark's, still a large speedup over sequential. 10 was chosen to bound concurrent in-flight requests on the shared ZK ensemble during leadership-acquisition bursts.

Testing Done

Unit tests in PR (all pass):

• TestCheckLiveInstancesObservationDeferred (3 tests) - verifies INIT defers, CALLBACK registers inline, take-and-clear semantics
• TestInitHandlersParallel (5 tests) - null/empty/single handler, parallel execution (verifies concurrency > 1), exception handling (one failure does not block others)

Existing integration tests (all pass, not modified):

• TestHandleSession (6 tests) - session re-establishment with real ZK
• TestDistControllerElection (4 tests) - controller leadership failover with real ZK
• TestZkCallbackHandlerLeak (5 tests) - no handler leaks after session expiry

---

Real-Cluster Validation (in addition to the mock benchmark)

Doc: https://docs.google.com/document/d/14LoOytLq4-YwwcxjRPCoxvWI6520kQj3JcviL5SYX8k/edit?usp=sharing

To address the ask for non-mock validation, the change was exercised end-to-end on a
real standalone ZooKeeper with a Helix distributed-controller ("supercluster")
topology — not in-memory ZK, not mocks.

Setup

• 1 real standalone ZooKeeper.
• 1 grand/controller cluster (SUPER_CLUSTER) running distributed controllers.
• 2 managed data clusters, each 100 live participants × 30 resources (64 partitions ×
  3 replicas), --activateCluster'd into the grand cluster.
• Cluster fully converged so CURRENTSTATES are populated, then measured on a clean
  single-controller failover (the scenario this PR targets: leadership acquisition over a
  large, already-populated cluster). 7,495 total ZK watches.

A/B method (apples-to-apples)
Same ZK, same participants, same resources — only the controller binary is swapped:

• BASELINE = PR base commit c901f6ea (serial registration).
• BRANCH = this PR (parallel registration).
  Metric is version-agnostic: wall-time between the first and last per-instance
  CallbackHandler subscribe for a cluster (this log line exists in both versions), so the
  comparison does not rely on a log line the PR itself adds.

Result — measured watch-setup (leadership-acquisition) time (reproduced across 2 runs)

Cluster	Baseline (serial)	This PR (parallel)	Speedup
Cluster A	5,458 ms	845 / 1,025 ms	5.3–6.5x
Cluster B	5,552 ms	988 / 977 ms	5.6–5.7x

(Each cluster: 400 per-instance subscribes = 100 instances × 4 listener types. Baseline
measured once on the identical populated cluster; PR measured on two separate runs to show
the result is stable.)

Mechanism confirmed on a live cluster

• Real path execution (no mocks): the live controller log shows the PR's own methods running —
  checkLiveInstancesObservation() deferring during INIT, registerPendingInstanceListeners()
  start/finish, and the dedicated registerInstanceListener-<cluster> threads doing the
  subscribes. These symbols are absent at PR base c901f6ea, so the path could only come from
  this PR's code.
• The controller registers exactly 4 listeners per instance (currentState, message,
  customizedStateRoot, taskCurrentState) → 100 × 4 = 400, matching the code's intent.
• The resulting ZK watch fan-out on the CURRENTSTATES subtree is
  #instances × (1 + #resources) = 100 × (1 + 30) = 3,100 per cluster (verified via ZK
  wchp) — i.e. the cost the serial loop pays grows with #instances × #resources,
  exactly the regime the change targets.
• No spurious "already initialized" initHandlers WARN at scale.

Note on magnitude vs the mock (19.9x): local ZK round-trips here are ~14 ms/subscribe,
so the serial loop is ~5.5 s and the measured speedup is ~6x. The mock used 200 ms/roundtrip
(production-representative) and showed ~20x. The real-cluster number is therefore a
conservative lower bound — in production, where ZK round-trips dominate and clusters are
larger (e.g. venice-5: 300 instances → ~1,200 registrations), the gain is larger and
matches the mock's order of magnitude.

[Sanju98]

How the number of threads on the client side will affect the zookeeper ensemble?
One single zookeeper-server instance can handle 15K connections on total.

[kabragaurav]

The 20 threads share one ZkClient (one ZK connection). No new connections are created. Updated the comment to say this.

[Sanju98]

The calling function is wrapped inside a synchronised block(ZKHelixManager.java) and this won't actually run in parallel. Can you confirm this?

[kabragaurav]

handler.init() -> invoke() acquires synchronized(_manager) which is the same lock. Fixed: synchronized(this) now only wraps the list copy. Parallel execution runs outside the lock.

[Sanju98]

Do the listeners have init-order dependency? or is it completely irrelevant?

[kabragaurav]

No order dependency. Each handler subscribes to a different ZK path (one per instance for CURRENTSTATES, TASKCURRENTSTATES, etc.). They do not depend on each other.

[Sanju98]

can we redesign this?
If one thread fails, the executor is doing a fullShutDownNow. This might lead to half-initialised handlers.

[kabragaurav]

shutdownNow() runs in the finally block after all future.get() calls complete. If one handler throws, the exception is caught and logged, and the loop continues to wait for remaining handlers. No handler is left half-initialized. shutdownNow() only cancels tasks on InterruptedException (when we break early), which is the right behavior - stop fast on interrupt.

[Sanju98]

can we add an integration test that adds/removes listeners during a session re-establishment, not mocks?

[Sanju98]

Do the listeners have init-order dependency? or is it completely irrelevant?

[kabragaurav]

can we add an integration test that adds/removes listeners during a session re-establishment, not mocks?

Already covered by the existing TestHandleSession.testConcurrentInitCallbackHandlers - it adds/removes listeners during session events with real in-memory ZK. We ran it and it passes (6/6). Also ran TestZkCallbackHandlerLeak (5/5) which verifies no handler leaks after session expiry.

[LZD-PratyushBhatt]

I don't think this touches the cold-acquisition path that actually drives the MissingTopState SLA-0, and the benchmark doesn't show that it does.

Our SLA-0 is on cold leadership acquisition (onBecomeLeaderFromStandby -> fresh HelixManager -> connect()). As the ss shows, handleNewSession runs handleNewSessionAsController() before initHandlers(). The total time (5-10 mins)(the log which shows "acquired leadership … took:..") is spent inside handleNewSessionAsController()

[kabragaurav]

Benchmark file (not committed - for local performance comparison only):

TestRegisterPendingListenersBenchmark.java

Results (100 instances, 200ms simulated ZK latency):

	Time	Speedup
Sequential (old)	40,692 ms	1.0x
Parallel (new)	2,048 ms	19.9x

To run: make registerPendingInstanceListeners package-private temporarily, then:

mvn test -pl helix-core -Dtest=TestRegisterPendingListenersBenchmark -DfailIfNoTests=false -am

[kabragaurav]

I don't think this touches the cold-acquisition path that actually drives the MissingTopState SLA-0

@LZD-PratyushBhatt Thanks for review. The earlier commits parallelized initHandlers() which runs AFTER handleNewSessionAsController() and is mostly a no-op (handlers are already initialized by then).

Fixed in the latest commits. The parallelization now targets the correct path:

1. handleNewSessionAsController() -> _leaderElectionHandler.init() -> invoke() -> acquireLeadership() -> addListenersToController() -> addLiveInstanceChangeListener() -> INIT callback -> checkLiveInstancesObservation()

2. checkLiveInstancesObservation() during INIT now defers per-instance listener registration (collects pending instances instead of calling addXxxListener in a loop).

3. After handleNewSessionAsController() returns and invoke() releases synchronized(_manager), handleNewSession() calls registerPendingInstanceListeners() which registers all per-instance listeners in parallel using a 20-thread pool.

The "acquired leadership ... took: Xms" log (DistributedLeaderElection.java line 120) will still reflect most of the time, but the per-instance registration that previously dominated it (~240s for 300 instances) now runs after it, in ~12s.

Benchmark (calls the real registerPendingInstanceListeners method with simulated 200ms ZK latency):

• Sequential (old): 40,639ms
• Parallel (new): 2,044ms
• Speedup: 19.9x

EI deployment needed for production validation of the "acquired leadership took" timing.

[kabragaurav]

can we add an integration test that adds/removes listeners during a session re-establishment, not mocks?

@Sanju98 Updated since the last reply. The PR now includes two things:

Unit tests in the PR (8 tests, all pass):

• TestCheckLiveInstancesObservationDeferred (3 tests) - verifies INIT defers per-instance registration, CALLBACK still registers inline, and take-and-clear works
• TestInitHandlersParallel (5 tests) - verifies parallel initHandlers() with concurrency check, exception isolation, null/empty/single edge cases

Existing integration tests with real ZK (all pass, not modified):

• TestHandleSession (6 tests) - session re-establishment including testConcurrentInitCallbackHandlers
• TestDistControllerElection (4 tests) - controller leadership failover
• TestZkCallbackHandlerLeak (5 tests) - no handler leaks after session expiry

Also, the code has changed significantly since your earlier inline comments. The parallelization was moved from initHandlers() (wrong path) to handleNewSession() -> registerPendingInstanceListeners() (correct path — after handleNewSessionAsController() returns and invoke() releases the lock). Details in the updated PR description.