[controller][test] Add durable store update callback

[misyel]

Problem Statement

The controller does not provide an extension point at the durable store-update execution boundary. Implementations that need to react to the final store state cannot participate in the existing admin-operation retry and checkpoint semantics without modifying the controller execution path.

Solution

Add a generic StoreUpdateHandler that is injectable through VeniceControllerContext and defaults to a no-op.

For parent-controller UPDATE_STORE operations, AdminExecutionTask invokes the handler with:

• A read-only snapshot of the final store state.
• An immutable set of config keys copied from the durable UPDATE_STORE message.

The config-key set remains stable across retries, so handlers can identify the requested changes without comparing pre-update and post-update snapshots. The callback runs after the metadata update succeeds and before the successful execution ID advances. Handler failures propagate, leaving the operation eligible for the admin channel's existing retry and restart behavior.

Existing constructor signatures delegate to the no-op handler, preserving compatibility for deployments and tests that do not configure the extension.

Code changes

• Added new code behind a config. If so list the config names and their default values in the PR description.
• Introduced new log lines.
  • Confirmed if logs need to be rate limited to avoid excessive logging.

No configuration or log changes are introduced. The handler defaults to StoreUpdateHandler.NO_OP.

Concurrency-Specific Checks

Both reviewer and PR author to verify

• Code has no race conditions or thread safety issues.
• Proper synchronization mechanisms (e.g., synchronized, RWLock) are used where needed.
• No blocking calls inside critical sections that could lead to deadlocks or performance degradation.
• Verified thread-safe collections are used (e.g., ConcurrentHashMap, CopyOnWriteArrayList).
• Validated proper exception handling in multi-threaded code to avoid silent thread termination.

The callback runs outside the store repository lock. Implementations may be invoked concurrently for different stores and receive read-only store snapshots and immutable config-key sets. Exceptions intentionally propagate so failed operations are not checkpointed.

How was this PR tested?

• Local code review completed.
• New unit tests added.
• New integration tests added.
• Modified or extended existing tests.
• Verified backward compatibility (if applicable).

The focused controller tests cover default and explicit handler injection, parent-only invocation, final-state delivery, immutable config-key delivery, ordering before checkpoint advancement, child-controller behavior, and failure propagation. They also verify that retries receive the same config-key set from the durable admin message.

The integration test injects separate handlers into a real parent-child controller topology, fails the parent handler's first callback attempt, and verifies that the admin operation retries and eventually delivers the final read-only store state with the same immutable config-key set. It also verifies that the child handler is never invoked.

Does this PR introduce any user-facing or breaking changes?

• No. You can skip the rest of this section.
• Yes. Clearly explain the behavior change and its impact.

🤖 Generated with GitHub Copilot CLI

[Copilot]

🟡 Not ready to approve

The new callback path performs store fetch/clone work even when the default NO_OP handler is used, introducing avoidable overhead in the common case.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

This review doesn't count toward merge requirements. Sign up for the private preview to control whether Copilot approvals count.

Pull request overview

Adds an injectable extension point for parent-controller UPDATE_STORE admin operations so external code can react to the final durable store state with the same retry/checkpoint semantics as the controller’s admin execution path.

Changes:

• Introduces StoreUpdateHandler (default NO_OP) and wires it through VeniceControllerContext → VeniceController → VeniceControllerService → admin consumer pipeline.
• Invokes the handler in AdminExecutionTask after a successful UPDATE_STORE metadata update and before advancing the successful execution ID (passing a ReadOnlyStore snapshot).
• Adds/updates controller tests to validate defaulting, injection, parent-only invocation, ordering, and failure propagation.

File summaries

File	Description
services/venice-controller/src/main/java/com/linkedin/venice/controller/StoreUpdateHandler.java	Adds the new handler interface and NO_OP default.
services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceControllerContext.java	Allows injecting StoreUpdateHandler via context builder with NO_OP default.
services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceController.java	Plumbs the handler from context into controller service creation.
services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceControllerService.java	Threads the handler into AdminConsumerService creation while preserving existing constructor compatibility.
services/venice-controller/src/main/java/com/linkedin/venice/controller/kafka/consumer/AdminConsumerService.java	Stores and passes the handler into AdminConsumptionTask (with backward-compatible constructor).
services/venice-controller/src/main/java/com/linkedin/venice/controller/kafka/consumer/AdminConsumptionTask.java	Stores and passes the handler into AdminExecutionTask (with backward-compatible constructor).
services/venice-controller/src/main/java/com/linkedin/venice/controller/kafka/consumer/AdminExecutionTask.java	Invokes the handler for parent UPDATE_STORE before checkpoint advancement using a read-only store snapshot.
services/venice-controller/src/test/java/com/linkedin/venice/controller/VeniceControllerContextTest.java	Verifies default and explicit handler injection in controller context.
services/venice-controller/src/test/java/com/linkedin/venice/controller/kafka/consumer/AdminExecutionTaskTest.java	Adds tests for handler invocation ordering, parent/child behavior, read-only snapshot, and failure propagation.

Review details

• Files reviewed: 9/9 changed files
• Comments generated: 1
• Review effort level: Low

We're testing this review assessment. Please use 👍 or 👎 to tell us if it's correct.

[Copilot]

🟡 Not ready to approve

The new callback path in AdminExecutionTask performs an avoidable extra cloneStore() (and can NPE if getStore returns null), which is easy to fix and improves robustness/performance.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

This review doesn't count toward merge requirements. Sign up for the private preview to control whether Copilot approvals count.

Review details

Comments suppressed due to low confidence (1)

services/venice-controller/src/main/java/com/linkedin/venice/controller/kafka/consumer/AdminExecutionTask.java:397

• AdminExecutionTask fetches the final store snapshot via admin.getStore(...).cloneStore() and then wraps it in ReadOnlyStore. In this code path admin.getStore() already returns a cloned/detached store (via the Helix store repositories), so the extra cloneStore() adds avoidable overhead and can become a triple-clone in the controller stack. Also, admin.getStore(...) may return null; today that would NPE on cloneStore() and obscure the failure reason.

    if (storeUpdated && isParentController) {
      Store finalStore = admin.getStore(clusterName, storeName).cloneStore();
      // Invoke before advancing checkpoints so callback failures leave the admin operation eligible for retry.
      storeUpdateHandler.handleStoreUpdate(clusterName, new ReadOnlyStore(finalStore));
    }

• Files reviewed: 11/11 changed files
• Comments generated: 0 new
• Review effort level: Lite

We're testing this review assessment. Please use 👍 or 👎 to tell us if it's correct.

[Copilot]

🟡 Not ready to approve

The default NO_OP path still performs an extra store fetch/clone on every parent UPDATE_STORE, which is an avoidable performance/operational regression.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

This review doesn't count toward merge requirements. Sign up for the private preview to control whether Copilot approvals count.

Review details

Comments suppressed due to low confidence (1)

services/venice-controller/src/main/java/com/linkedin/venice/controller/kafka/consumer/AdminExecutionTask.java:401

• The store update callback path currently fetches and clones the store even when the injected handler is the default StoreUpdateHandler.NO_OP, which adds an extra metadata read and cloneStore() cost to every parent UPDATE_STORE operation even when no extension is configured. This can be a noticeable regression since the default behavior should be truly no-op.

    if (storeUpdated && isParentController) {
      Store finalStore = admin.getStore(clusterName, storeName).cloneStore();
      // Invoke before advancing checkpoints so callback failures leave the admin operation eligible for retry.
      storeUpdateHandler.handleStoreUpdate(clusterName, new ReadOnlyStore(finalStore), updatedConfigs);
    }

• Files reviewed: 11/11 changed files
• Comments generated: 0 new
• Review effort level: Lite

We're testing this review assessment. Please use 👍 or 👎 to tell us if it's correct.

[Copilot]

🟡 Not ready to approve

The new callback path dereferences admin.getStore(...).cloneStore() without a null-check, risking an NPE with an unclear failure mode during admin-op processing.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

This review doesn't count toward merge requirements. Sign up for the private preview to control whether Copilot approvals count.

Review details

Suppressed comments (1)

services/venice-controller/src/main/java/com/linkedin/venice/controller/kafka/consumer/AdminExecutionTask.java:401

• admin.getStore(clusterName, storeName) can return null (e.g., store deleted or metadata temporarily unavailable). Calling .cloneStore() without a null-check will throw an NPE and obscure the actual failure; it’s better to fail with an explicit exception so the admin op retries with a clear reason.

    if (storeUpdated && isParentController && !storeUpdateHandler.isNoOp()) {
      Store finalStore = admin.getStore(clusterName, storeName).cloneStore();
      // Invoke before advancing checkpoints so callback failures leave the admin operation eligible for retry.
      storeUpdateHandler.handleStoreUpdate(clusterName, new ReadOnlyStore(finalStore), updatedConfigs);
    }

• Files reviewed: 11/11 changed files
• Comments generated: 0 new
• Review effort level: Lite

We're testing this review assessment. Please use 👍 or 👎 to tell us if it's correct.