[controller] Remove topic-based concurrent push detection and clarify push-blocking messaging

[KaiSernLim]

Problem Statement

Two related issues in the parent controller's concurrent-push detection:

1. Dead flag. ConcurrentPushDetectionStrategy (concurrent.push.detection.strategy) gated two ways of detecting an in-flight push: the legacy "topic-based" path (derives the current push from parent version topics) and the newer "parent-version-status" path. Production parent controllers have fully migrated to PARENT_VERSION_STATUS_ONLY, so the topic-based path and the flag are dead in production, yet they still carried significant code (the parent creating/truncating version topics, orphan-topic cleanup, etc.). A TODO in VeniceController called for removing it once the new mode was fully rolled out.

2. Ambiguous push-blocking message. When a new push is rejected because a prior version exists, the rejection message branched only on version.isVersionSwapDeferred(). A deferred-swap-enabled version whose push was still in progress was therefore reported as "make that version current" — implying it only needed a roll-forward. This ambiguity sent an investigation down the wrong path on an issue that was actually a concurrent push, not a deferred (colo-by-colo) version-swap wait.

Solution

Remove topic-based tracking; make version-status tracking the only path:

• getTopicForCurrentPushJob always delegates to the parent-version-status path. Deletes getTopicForCurrentPushJobTopicBasedTracking and the now-orphaned helpers getKafkaTopicsByAge / truncateTopicsBasedOnMaxErroredTopicNumToKeep.
• Removes the now-dead isTopicWriteNeeded() gates: the parent no longer creates or truncates the parent version topic (addVersion, shouldSkipTruncatingTopic, rollForwardToFutureVersion, the push-completion path, killOfflinePush, DeferredVersionSwapService).
• Deletes the ConcurrentPushDetectionStrategy enum, the concurrent.push.detection.strategy config key, and the cluster-config field/parse/getter.
• Behavior-preserving for LinkedIn (prod parents already run PARENT_VERSION_STATUS_ONLY); it changes the OSS default, which was DUAL.
• DeferredVersionSwapService (controller.deferred.version.swap.service.enabled) is a separate, prod-enabled feature and is intentionally kept.

Clarify the push-blocking message:

• ConcurrentBatchPushException now distinguishes the cases by the version's actual status: a deferred version that is PUSHED (push complete, swap not yet done) but not yet current in all regions is reported as waiting on deferred version swap (with per-region current versions and the target swap region); everything else is reported as an in-progress concurrent push (including the version status).
• validateChildCurrentVersions logs the full per-region picture and names the deferred (colo-by-colo) version swap, and takes the Version so its status is included in the log line.

Version-status-based push-job topic lookup. getTopicForCurrentPushJob branches on the latest version's status:

• Terminal statuses (ONLINE, ERROR, KILLED, ROLLED_BACK, PARTIALLY_ONLINE) and NOT_CREATED — no ongoing push, never block. NOT_CREATED is included because VersionStatus documents it as the safe rolling-deployment fallback for an unrecognized status id that must not block new pushes. (ONLINE also fixes a prior stale-store misdetection.)
• A non-deferred STARTED version that is already current in the store / serving in all regions has effectively completed and is allowed to proceed.
• Everything else — CREATED, in-flight STARTED, PUSHED, and any deferred-swap version not yet current in every region — blocks, because a future version still occupies the store. In particular a STARTED version blocks unconditionally rather than polling child job status: job-status polling only confirms ingestion completion, which for a deferred-swap version does not imply the version swap is done, so polling would open a race where a concurrent push slips in during the STARTED → PUSHED transition.
• isIncrementalPush/isRepush are accepted for call-site/interface compatibility but are unused by this parent-only implementation (documented in the Javadoc).

Parent-fabric topic cleanup moved into the base TopicCleanupService. With the parent no longer relying on topic-based tracking, TopicCleanupServiceForParentController was removed and VeniceController always constructs the base TopicCleanupService. Because the parent still truncates topics across every parent fabric (VeniceHelixAdmin#truncateKafkaTopicInParentFabrics), the base service now runs cleanup against each parent fabric's Kafka cluster when admin.isParent() (and the single default cluster otherwise), preserving the old subclass's multi-fabric behavior:

• getTopicManagersForCleanup() returns one TopicManager per configured parent fabric, skipping any fabric missing a getChildDataCenterKafkaUrlMap() entry (with a warning) instead of crashing on getTopicManager(null), and falling back to the default manager if none are configured.
• The version-topic deletion delay countdown key now includes the Kafka cluster address, so each fabric keeps an independent countdown (matching the removed subclass) instead of being decremented once per fabric per cycle.
• Also inlined the now-trivial shouldSkipTruncatingTopic(clusterName) (return isParent()) at its one call site.

Code changes

• Removed a config (concurrent.push.detection.strategy); no new config added.
• Changed log lines (validateChildCurrentVersions, getTopicForCurrentPushJob); volume is unchanged (at most one line per blocked push), so no new rate-limiting is needed.

Concurrency-Specific Checks

• Removes dead branches/services and rewords messages. The one behavioral concurrency change is that an in-flight STARTED version now blocks the next push unconditionally (closing the STARTED → PUSHED swap race for deferred-swap versions) rather than being unblocked by ingestion-only job-status polling.

How was this PR tested?

• New unit tests added (testDeferredVersionSwapWaitMessageIsDistinctFromConcurrentPush, testGetTopicManagersForCleanup, NOT_CREATED coverage in testGetTopicForCurrentPushJob).
• Modified or extended existing tests (testGetTopicForCurrentPushJobBlocksInProgressVersion).
• Unit: TestVeniceParentHelixAdmin, TestVeniceHelixAdmin, TestDeferredVersionSwapService, TestDeferredVersionSwapServiceWithSequentialRollout, TestTopicCleanupService — all pass.
• Integration: TestDeferredVersionSwap and TestTargetedRegionPushWithNativeReplication — all pass (incl. target-region deferred swap). TestParentControllerWithMultiDataCenter assertions were updated to the new wording.
• spotbugsMain (venice-controller + venice-common) and spotlessApply are clean.

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

• Yes.
  • OSS default change: default concurrent-push detection changes from DUAL (ran both paths, returned the topic-based result) to parent-version-status-only. The concurrent.push.detection.strategy config is removed and ignored if still set. Behavior is equivalent for the supported deferred-swap and concurrent-push scenarios (validated by the deferred-swap / target-region integration tests); LinkedIn production already ran the version-status path.
  • Message change: the ConcurrentBatchPushException text shown to push jobs now distinguishes "waiting on deferred version swap" (with per-region status) from "ongoing concurrent push" (with the version status).
  • Parent controller no longer runs TopicCleanupServiceForParentController: it now uses the same base TopicCleanupService as child controllers, which iterates all parent fabrics' Kafka clusters when running on a parent. Behavior for operators is unchanged.

🤖 Generated with Claude Code

[Copilot]

Pull request overview

This PR removes the legacy/topic-based concurrent push detection path from the parent controller and standardizes on parent-version-status-based tracking, while also improving the push-blocking messaging to clearly distinguish “in-progress concurrent push” vs “waiting on deferred version swap”.

Changes:

• Remove ConcurrentPushDetectionStrategy and the concurrent.push.detection.strategy config, along with topic-based current-push detection and related topic truncation gates on the parent controller.
• Update parent/controller logic and tests to reflect that parent controllers no longer create/truncate version topics (and simplify shouldSkipTruncatingTopic accordingly).
• Clarify ConcurrentBatchPushException messaging and logging to explicitly call out deferred version swap waits (with per-region current versions and target swap region).

Reviewed changes

Copilot reviewed 13 out of 13 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceParentHelixAdmin.java	Removes topic-based detection path; updates push-blocking message; removes parent VT truncation gates.
services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceHelixAdmin.java	Stops parent controllers from creating version topics; simplifies truncation-skip logic for parents.
services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceControllerClusterConfig.java	Removes parsing/storage of the concurrent push detection strategy config.
services/venice-controller/src/main/java/com/linkedin/venice/controller/DeferredVersionSwapService.java	Removes parent VT truncation behavior from deferred swap promotion.
services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceController.java	Updates TODO/commentary now that parent no longer writes version topics.
internal/venice-common/src/main/java/com/linkedin/venice/meta/ConcurrentPushDetectionStrategy.java	Deletes the enum (strategy fully removed).
internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java	Removes the concurrent.push.detection.strategy key.
services/venice-controller/src/test/java/com/linkedin/venice/controller/TestVeniceParentHelixAdmin.java	Updates tests for new behavior and adds coverage for distinct deferred-swap vs concurrent-push messaging.
services/venice-controller/src/test/java/com/linkedin/venice/controller/TestVeniceHelixAdmin.java	Simplifies truncation-skip tests now that parent always skips.
services/venice-controller/src/test/java/com/linkedin/venice/controller/AbstractTestVeniceParentHelixAdmin.java	Removes now-dead strategy stubbing.
services/venice-controller/src/test/java/com/linkedin/venice/controller/TestDeferredVersionSwapService.java	Removes now-dead strategy stubbing.
services/venice-controller/src/test/java/com/linkedin/venice/controller/TestDeferredVersionSwapServiceWithSequentialRollout.java	Removes now-dead strategy stubbing and truncation assertions.
internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestParentControllerWithMultiDataCenter.java	Updates assertions for the clarified push-blocking message text.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated 2 comments.

[misyel]

Why did we remove the STARTED status check?

[KaiSernLim]

🤖 Good catch — STARTED was removed from this block intentionally, but the reason wasn't spelled out clearly. A STARTED version may already be terminal in child regions but not yet reflected on the parent (the parent only transitions out of STARTED once a job-status poll observes a terminal child status). Polling PUSHED children, on the other hand, would always return terminal and incorrectly unblock the next push. So STARTED is handled by the polling branch further below, while CREATED and PUSHED block unconditionally here. Added a comment to the code to make this explicit.

[misyel]

I think polling job status based on STARTED version status can allow concurrent push to happen because polling on job status will only verify that ingestion is completed and doesn't check that version swap is complete too. There will be a small race window where concurrent push can happen between started -> pushed version transition if we want to poll job status for started version statuses. Can you check if that is true?

[KaiSernLim]

🤖 You're right, and good catch. Verified: the STARTED-polling branch only observed ingestion completion via getOffLinePushStatus, which for a deferred-swap version does not imply the version swap is done — so a terminal poll could unblock the next push during the STARTED → PUSHED transition, exactly the race you described.

Fixed in ff18c59b7: a STARTED latest version now blocks the next push unconditionally, the same as CREATED and PUSHED, and I removed the polling loop entirely. This matches the getTopicForCurrentPushJobParentVersionStatusBasedTracking behavior on main (STARTED/PUSHED/CREATED all block; the poll below it there is effectively unreachable). Updated testGetTopicForCurrentPushJobBlocksInProgressVersion to assert all three non-terminal statuses block immediately without polling.

[misyel]

Should we check for PUSHED instead? ONLINE means ingestion and version swap completed so the log below will never log

[KaiSernLim]

🤖 Yes, good catch. Changed to PUSHED. PUSHED is the parent-side status when a deferred-swap version has completed its push but its version swap is still pending — i.e., when a new push should be blocked. ONLINE on the parent for a deferred-swap version means the swap has already completed (and validateChildCurrentVersions would have unblocked the push upstream), so the ONLINE branch in incrementVersionIdempotent would never actually be reached. Fixed and re-added the corresponding test (testDeferredVersionSwapWaitMessageIsDistinctFromConcurrentPush) with VersionStatus.PUSHED.

[Copilot]

Pull request overview

Copilot reviewed 16 out of 18 changed files in this pull request and generated 2 comments.

Files not reviewed (1)

• .impeccable/hook.cache.json: Generated file

[Copilot]

Pull request overview

Copilot reviewed 16 out of 16 changed files in this pull request and generated 2 comments.

[github-actions]

Hi there. This pull request has been inactive for 30 days. To keep our review queue healthy, we plan to close it in 7 days unless there is new activity. If you are still working on this, please push a commit, leave a comment, or convert it to draft to signal intent. Thank you for your time and contributions.

[github-actions]

Closing this pull request due to 37 days of inactivity. This is not a judgment on the value of the work. If you would like to continue, please reopen or open a new PR and we will be happy to take another look. Thank you again for contributing.

[Copilot]

Pull request overview

Copilot reviewed 16 out of 16 changed files in this pull request and generated no new comments.

Suppressed comments (1)

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceParentHelixAdmin.java:3694

• In truncateTopicsOptionally, the maxErroredTopicNumToKeep>0/error branch still appends/logs "Parent Kafka topic won't be truncated". After removing parent version-topic truncation, this status detail is misleading (the only truncation that can happen here now is for the stream-reprocessing topic). Consider rewording the status detail/log to avoid implying the parent version topic would otherwise be truncated.

      boolean isTargetRegionPushWithDeferredSwap =
          isDeferredVersionSwap && !StringUtils.isEmpty(version.getTargetSwapRegion());

      if ((failedBatchPush || nonIncPushBatchSuccess && !isDeferredVersionSwap || incPushEnabledBatchPushSuccess

[Copilot]

Pull request overview

Copilot reviewed 16 out of 16 changed files in this pull request and generated 2 comments.

Suppressed comments (1)

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceParentHelixAdmin.java:1420

• getTopicForCurrentPushJob treats any non-terminal status as blocking, but VersionStatus.NOT_CREATED is explicitly documented as an inert fallback for unknown status ids during rolling deploys (it “won't block new pushes”). With the current switch, a NOT_CREATED latest version will block indefinitely and return a version topic.

Consider explicitly treating NOT_CREATED as a non-blocking status here (similar to other terminal statuses) to avoid stuck pushes when the parent reads an unrecognized status value.

    switch (lastVersion.getStatus()) {
      case KILLED:
      case ERROR:
      case ROLLED_BACK:
      case PARTIALLY_ONLINE:

[Copilot]

Pull request overview

Copilot reviewed 16 out of 16 changed files in this pull request and generated 1 comment.

Suppressed comments (1)

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceParentHelixAdmin.java:1423

• getTopicForCurrentPushJob currently treats VersionStatus.NOT_CREATED as a non-terminal status and will block the next push. However, VersionStatus.getVersionStatusFromInt intentionally maps unknown/new status IDs to NOT_CREATED as the safest inert fallback specifically so older controllers do not block new pushes during rolling deployments (see internal/venice-common/.../VersionStatus.java). Treating NOT_CREATED as blocking here could cause unnecessary push rejections in mixed-version windows.

    // CREATED/PUSHED block the next push outright. STARTED can still represent a normal push whose
    // current-version promotion has already completed but whose parent version status has not caught up.
    switch (lastVersion.getStatus()) {
      case KILLED:
      case ERROR:
      case ROLLED_BACK:
      case PARTIALLY_ONLINE:
      case ONLINE:
        LOGGER.info(

[Copilot]

Pull request overview

Copilot reviewed 16 out of 16 changed files in this pull request and generated no new comments.

Suppressed comments (3)

Previously missed (1) — in code that hasn't changed since the last review.

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceParentHelixAdmin.java:1444

• In getTopicForCurrentPushJob, the STARTED/non-deferred branch always calls getCurrentVersionsForMultiColos even when store.getCurrentVersion() == lastVersionNum would already allow the push to proceed. getCurrentVersionsForMultiColos can be relatively expensive (fan-out to child controllers), so this adds unnecessary remote work for the common “already current but status still STARTED” case. Consider short-circuiting on store.getCurrentVersion() before fetching per-region current versions.

    if (lastVersion.getStatus() == STARTED && !lastVersion.isVersionSwapDeferred()) {
      Map<String, Integer> currentVersions = getCurrentVersionsForMultiColos(clusterName, storeName);
      boolean allFabricsServingLastVersion = currentVersions != null && !currentVersions.isEmpty()
          && currentVersions.values()
              .stream()
              .allMatch(currentVersion -> Objects.equals(currentVersion, lastVersionNum));
      if (store.getCurrentVersion() == lastVersionNum || allFabricsServingLastVersion) {
        LOGGER.info(

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceParentHelixAdmin.java:1465

• PR description states that STARTED versions are polled via child job status, but the final getTopicForCurrentPushJob implementation explicitly avoids polling (no getOffLinePushStatus path) and always blocks in non-terminal states unless the version is already current. Please align the PR description/commit message with the implemented behavior to avoid confusion for reviewers/operators.

    // push completed in its target region but whose swap is still pending). In all of these the version
    // is not yet done from the user's perspective, so the next push must wait. STARTED is treated the
    // same as CREATED/PUSHED for deferred-swap versions, or when the version is not current yet, rather
    // than polling the child job status: a job-status poll only observes ingestion completion, which for
    // a deferred-swap version does not imply the version swap has finished, so unblocking on a terminal
    // poll would let a concurrent push slip in during the STARTED -> PUSHED transition.
    LOGGER.info(

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceController.java:301

• createTopicCleanupService now always wires the base TopicCleanupService, but TopicCleanupService only enumerates topics via admin.getTopicManager() (local cluster) and does not iterate multiClusterConfigs.getParentFabrics(). Previously, the parent-specific service cleaned deprecated topics across all parent-fabric Kafka clusters. If parentFabrics is non-empty in a deployment, this change can leave deprecated topics in non-local parent fabrics uncleaned, which may cause operational topic leaks.

  private TopicCleanupService createTopicCleanupService() {
    Admin admin = controllerService.getVeniceHelixAdmin();
    return new TopicCleanupService(
        admin,
        multiClusterConfigs,
        pubSubTopicRepository,
        new TopicCleanupServiceStats(metricsRepository),
        pubSubClientsFactory);

[Copilot]

Pull request overview

Copilot reviewed 18 out of 18 changed files in this pull request and generated 2 comments.

Suppressed comments (1)

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceParentHelixAdmin.java:1464

• The PR description says the parent will "Poll child status for STARTED versions", but the final getTopicForCurrentPushJob implementation does not poll child offline-push status at all (it explicitly blocks without polling). Please either update the PR description to match the current behavior, or reintroduce the STARTED-only polling behavior if that was a requirement.

    // The only statuses left after the terminal cases above are non-terminal: CREATED (version exists
    // but its push has not begun), STARTED (push in flight), and PUSHED (a deferred-swap version whose
    // push completed in its target region but whose swap is still pending). In all of these the version
    // is not yet done from the user's perspective, so the next push must wait. STARTED is treated the
    // same as CREATED/PUSHED for deferred-swap versions, or when the version is not current yet, rather
    // than polling the child job status: a job-status poll only observes ingestion completion, which for
    // a deferred-swap version does not imply the version swap has finished, so unblocking on a terminal
    // poll would let a concurrent push slip in during the STARTED -> PUSHED transition.

[Copilot]

Pull request overview

Copilot reviewed 18 out of 18 changed files in this pull request and generated 1 comment.

Suppressed comments (1)

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceParentHelixAdmin.java:1464

• PR description says STARTED versions “poll the child job status”, but the current implementation explicitly does not poll (it blocks based on parent version status and current-version checks only). Please update the PR description (and/or any commit message) to match the final behavior so reviewers/operators aren’t misled about how concurrent-push detection works now.

    // is not yet done from the user's perspective, so the next push must wait. STARTED is treated the
    // same as CREATED/PUSHED for deferred-swap versions, or when the version is not current yet, rather
    // than polling the child job status: a job-status poll only observes ingestion completion, which for
    // a deferred-swap version does not imply the version swap has finished, so unblocking on a terminal
    // poll would let a concurrent push slip in during the STARTED -> PUSHED transition.

[Copilot]

Pull request overview

Copilot reviewed 18 out of 18 changed files in this pull request and generated no new comments.