[dvc][cc] Fix unsafe DVRT CDC version swap

.github/workflows/VeniceCI-E2ETests.yml

@@ -1629,6 +1629,44 @@ jobs:
         with:
           shard: 85
 
+  IntegrationTests_86:
+    name: IntegrationTests_86
+    strategy:
+      fail-fast: false
+    runs-on: ubuntu-latest
+    permissions:
+     contents: read
+    timeout-minutes: 15
+    concurrency:
+     group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}-jdk17-IntegrationTests_86
+     cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+    steps:
+      - uses: actions/checkout@v6
+        with:
+          fetch-depth: 0
+      - uses: ./.github/actions/integration-test-run
+        with:
+          shard: 86
+
+  IntegrationTests_87:
+    name: IntegrationTests_87
+    strategy:
+      fail-fast: false
+    runs-on: ubuntu-latest
+    permissions:
+     contents: read
+    timeout-minutes: 15
+    concurrency:
+     group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}-jdk17-IntegrationTests_87
+     cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+    steps:
+      - uses: actions/checkout@v6
+        with:
+          fetch-depth: 0
+      - uses: ./.github/actions/integration-test-run
+        with:
+          shard: 87
+
   IntegrationTests_99:
     name: IntegrationTests_99
     strategy:
@@ -1656,7 +1694,7 @@ jobs:
     runs-on: ubuntu-latest
     permissions:
       contents: read
-    needs: [IntegrationTests_1, IntegrationTests_2, IntegrationTests_3, IntegrationTests_4, IntegrationTests_5, IntegrationTests_6, IntegrationTests_7, IntegrationTests_8, IntegrationTests_9, IntegrationTests_10, IntegrationTests_11, IntegrationTests_12, IntegrationTests_13, IntegrationTests_14, IntegrationTests_15, IntegrationTests_16, IntegrationTests_17, IntegrationTests_18, IntegrationTests_19, IntegrationTests_20, IntegrationTests_21, IntegrationTests_22, IntegrationTests_23, IntegrationTests_24, IntegrationTests_25, IntegrationTests_26, IntegrationTests_27, IntegrationTests_28, IntegrationTests_29, IntegrationTests_30, IntegrationTests_31, IntegrationTests_32, IntegrationTests_33, IntegrationTests_34, IntegrationTests_35, IntegrationTests_36, IntegrationTests_37, IntegrationTests_38, IntegrationTests_39, IntegrationTests_40, IntegrationTests_41, IntegrationTests_42, IntegrationTests_43, IntegrationTests_44, IntegrationTests_45, IntegrationTests_46, IntegrationTests_47, IntegrationTests_48, IntegrationTests_49, IntegrationTests_50, IntegrationTests_51, IntegrationTests_52, IntegrationTests_53, IntegrationTests_54, IntegrationTests_55, IntegrationTests_56, IntegrationTests_57, IntegrationTests_58, IntegrationTests_59, IntegrationTests_60, IntegrationTests_61, IntegrationTests_62, IntegrationTests_63, IntegrationTests_64, IntegrationTests_65, IntegrationTests_66, IntegrationTests_67, IntegrationTests_68, IntegrationTests_69, IntegrationTests_70, IntegrationTests_71, IntegrationTests_72, IntegrationTests_73, IntegrationTests_74, IntegrationTests_75, IntegrationTests_76, IntegrationTests_77, IntegrationTests_78, IntegrationTests_79, IntegrationTests_80, IntegrationTests_81, IntegrationTests_82, IntegrationTests_83, IntegrationTests_84, IntegrationTests_85, IntegrationTests_99]
+    needs: [IntegrationTests_1, IntegrationTests_2, IntegrationTests_3, IntegrationTests_4, IntegrationTests_5, IntegrationTests_6, IntegrationTests_7, IntegrationTests_8, IntegrationTests_9, IntegrationTests_10, IntegrationTests_11, IntegrationTests_12, IntegrationTests_13, IntegrationTests_14, IntegrationTests_15, IntegrationTests_16, IntegrationTests_17, IntegrationTests_18, IntegrationTests_19, IntegrationTests_20, IntegrationTests_21, IntegrationTests_22, IntegrationTests_23, IntegrationTests_24, IntegrationTests_25, IntegrationTests_26, IntegrationTests_27, IntegrationTests_28, IntegrationTests_29, IntegrationTests_30, IntegrationTests_31, IntegrationTests_32, IntegrationTests_33, IntegrationTests_34, IntegrationTests_35, IntegrationTests_36, IntegrationTests_37, IntegrationTests_38, IntegrationTests_39, IntegrationTests_40, IntegrationTests_41, IntegrationTests_42, IntegrationTests_43, IntegrationTests_44, IntegrationTests_45, IntegrationTests_46, IntegrationTests_47, IntegrationTests_48, IntegrationTests_49, IntegrationTests_50, IntegrationTests_51, IntegrationTests_52, IntegrationTests_53, IntegrationTests_54, IntegrationTests_55, IntegrationTests_56, IntegrationTests_57, IntegrationTests_58, IntegrationTests_59, IntegrationTests_60, IntegrationTests_61, IntegrationTests_62, IntegrationTests_63, IntegrationTests_64, IntegrationTests_65, IntegrationTests_66, IntegrationTests_67, IntegrationTests_68, IntegrationTests_69, IntegrationTests_70, IntegrationTests_71, IntegrationTests_72, IntegrationTests_73, IntegrationTests_74, IntegrationTests_75, IntegrationTests_76, IntegrationTests_77, IntegrationTests_78, IntegrationTests_79, IntegrationTests_80, IntegrationTests_81, IntegrationTests_82, IntegrationTests_83, IntegrationTests_84, IntegrationTests_85, IntegrationTests_86, IntegrationTests_87, IntegrationTests_99]
     timeout-minutes: 20
     if: ${{ cancelled() || contains(needs.*.result, 'cancelled') || contains(needs.*.result, 'failure') }}
     steps:

AA_DVRT_CDC_VERSION_SWAP_PLAN.md

@@ -0,0 +1,125 @@
+# AA DVRT CDC Version-Swap Coordination — Plan & Design
+
+Working notes for PR [#2795 "[dvc][cc] Fix unsafe DVRT CDC version swap"](https://github.com/linkedin/venice/pull/2795).
+This doc captures the problem, design, and remaining work so the context survives across sessions. It is a planning
+artifact, not user-facing documentation — it can be dropped in a pre-merge cleanup.
+
+## Status
+
+- **PR:** #2795 (open, not draft) against `linkedin/venice`, head branch `worktree-dvrt-cdc-aa-version-swap`.
+- **Mergeability:** no git conflicts, but **BLOCKED** on CI.
+- **CI:** `IntegrationTests_67` and `IntegrationTests_87` failing; `E2ETestsFailureAlert` fires as the aggregate of
+  those two. ~110 other checks green.
+- **Review:** no maintainer decision yet. Only open threads are from the Copilot bot.
+- **Last move:** HEAD reverts an earlier "tighten swap-settle window to cut shard runtime" test-timing tweak — the AA
+  swap integration tests were being tuned for shard runtime/flakiness when work paused.
+
+## Problem
+
+The DVRT-based CDC consumer's per-partition version-swap flip is unsafe in three ways — the same three issues PR #2280
+fixed on the legacy `VeniceChangelogConsumerImpl`, layered on top of the #2245 server-side multi-region
+VersionSwapMessage (VSM) broadcast:
+
+1. **Acts on VSMs from another region.** Under active-active the same logical swap is broadcast once per source region.
+   The client must act only on VSMs whose `sourceRegion` matches its own region.
+2. **Acts on stale VSMs from a previous version.** Re-pushes replay prior VTs' VSMs into the new VT;
+   restart-from-EARLIEST and rollback can also replay historical VSMs. Must filter by `generationId` +
+   `oldServingVersionTopic`/`newServingVersionTopic` and reject any swap targeting a version not greater than the
+   highest already served (and reject `NON_EXISTING_VERSION`).
+3. **Future-version ingestion outpaces current-version ingestion → data loss at swap.** The current-version transformer
+   is throttled by the user's post-poll processing (records drain through `poll()` before more are ingested), but the
+   future-version transformer's ingestion bypasses user post-poll. Whenever user post-poll is slow (the common case, not
+   just A/A cross-DC skew) the future side can race past the swap point before the current side reaches it, silently
+   swallowing in-flight records that should have surfaced on the current side.
+
+All three are concrete safety bugs documented in #2245 and #2280. Issue 3 is more general than A/A — it is a property of
+how DVRT CDC interleaves user post-poll back-pressure with version-specific ingestion.
+
+## Design
+
+New class `RecordTransformerVersionSwapCoordinator`
+(`clients/da-vinci-client/src/main/java/com/linkedin/davinci/consumer/`). Owned by
+`VeniceChangelogConsumerDaVinciRecordTransformerImpl` when
+`ChangelogClientConfig.isVersionSwapByControlMessageEnabled()` is true.
+
+- **Barrier.** Enforces a cross-partition, cross-region barrier: a swap commits only after every assigned partition has
+  observed VSMs from every region on **both** the current and future version topics. Per-partition × per-region
+  observations are accumulated separately for the current (`currentVersionRegionsConsumed`) and future
+  (`futureVersionRegionsConsumed`) sides.
+- **State machine.** `IDLE → IN_PROGRESS → {COMMITTED, TIMED_OUT, FAILED} → IDLE`. Per-swap state (generationId, old/new
+  VT, target version, the max-served snapshot, per-side region sets, paused partition sets) is reset on arm and after
+  any terminal state.
+- **VSM gating (`isRelevant`).** Rejects a VSM unless: `generationId != -1`; `sourceRegion` equals the client's region;
+  both old/new serving VTs are present and the new one is a valid version topic; the topic matches this side (old for
+  current, new for future); the target version is greater than `NON_EXISTING_VERSION` and greater than the arm-time
+  max-served snapshot; and, while `IN_PROGRESS`, the `generationId` and old/new topics match the active swap.
+  `maxServedVersion` is snapshotted once at arm-time so gating stays O(partitions), not O(partitions²).
+- **Cutover / pause.** Once a partition's per-side barrier closes, Kafka prefetch is paused on that side.
+  `consumer.pause()` does not truncate the in-flight batch, so records past the VSM in the same poll-batch still flow
+  through `processPut`: surfaced on the current side, dropped on the future side.
+- **Commit.** When the barrier closes for all assigned partitions, `partitionToVersionToServe` is flipped atomically for
+  all of them inside the synchronized block, then future-side prefetch resumes.
+- **Timeout watchdog.** A daemon `ScheduledExecutorService` force-commits (counts as SUCCESS, not failure) if the
+  barrier does not close within `ChangelogClientConfig.setVersionSwapTimeoutInMs(...)`, bounding the vulnerable cutover
+  window.
+- **Mid-swap unsubscribe.** Removing a partition drops it from the barrier; if that unblocks the remaining set, the swap
+  commits automatically.
+- **Thread safety.** Every state-mutating method is `synchronized` on the instance; the watchdog timer contends for the
+  same lock.
+
+### Wiring
+
+- `VeniceChangelogConsumerDaVinciRecordTransformerImpl.onVersionSwap()` has two branches — AA (coordinator) and legacy
+  per-partition flip — selected by `isVersionSwapByControlMessageEnabled()`. The legacy path (`false`) must be preserved
+  exactly.
+- `InternalDaVinciRecordTransformer` gains the pause/resume partition-consumption hooks and a back-reference lifecycle
+  hook so the coordinator can pause/resume the right side's Kafka prefetch.
+- `StoreIngestionTask` wires the pause/resume lambdas to the version's Kafka consumer service at transformer
+  construction and sets the back-reference post-construction.
+
+## Known tradeoff
+
+The coordinator intentionally allows record loss inside the cutover window, under the assumption that current and future
+leaders process RT events in approximately aligned poll-batches. This is documented in the coordinator's class javadoc
+and asserted in the AA integration tests. The timeout watchdog bounds the window. Any change to swap behavior must keep
+the legacy path (`isVersionSwapByControlMessageEnabled == false`) untouched and respect the coordinator's at-least-once
+guarantee scope.
+
+## Files in the change
+
+Main:
+
+- `clients/da-vinci-client/src/main/java/com/linkedin/davinci/consumer/RecordTransformerVersionSwapCoordinator.java`
+  (new)
+- `clients/da-vinci-client/src/main/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerDaVinciRecordTransformerImpl.java`
+- `clients/da-vinci-client/src/main/java/com/linkedin/davinci/client/InternalDaVinciRecordTransformer.java`
+- `clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java`
+
+Tests:
+
+- `clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/RecordTransformerVersionSwapCoordinatorTest.java`
+  (new)
+- `clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerDaVinciRecordTransformerAaVersionSwapTest.java`
+  (new)
+- `clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerDaVinciRecordTransformerImplTest.java`
+- `clients/da-vinci-client/src/test/java/com/linkedin/davinci/transformer/RecordTransformerTest.java`
+- `internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/AbstractDvrtCdcAaVersionSwapTest.java`
+  (new — shared base)
+- `internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestDvrtCdcAaVersionSwap.java`
+  (new)
+- `internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestDvrtCdcAaVersionSwapMultiVersion.java`
+  (new)
+
+CI / infra:
+
+- `.github/workflows/VeniceCI-E2ETests.yml`
+- `internal/venice-test-common/test-shard-assignments.json`
+
+## Remaining work
+
+1. **Unblock CI.** Diagnose `IntegrationTests_67` and `IntegrationTests_87` — determine whether the failures are from
+   this change or timing/flakiness in the AA swap e2e tests (the reverted swap-settle tweak suggests the latter was in
+   play). Fix or re-stabilize, then confirm green.
+2. **Get a maintainer review.** `reviewDecision` is empty; only Copilot has commented. Resolve the remaining bot threads
+   and request a human reviewer.
+3. **Pre-merge cleanup.** Remove this planning doc (or move it out of the diff) before merge.

clients/da-vinci-client/src/main/java/com/linkedin/davinci/client/InternalDaVinciRecordTransformer.java

@@ -5,6 +5,7 @@
 import com.linkedin.venice.annotation.Experimental;
 import com.linkedin.venice.compression.VeniceCompressor;
 import com.linkedin.venice.kafka.protocol.ControlMessage;
+import com.linkedin.venice.kafka.protocol.VersionSwap;
 import com.linkedin.venice.kafka.protocol.state.PartitionState;
 import com.linkedin.venice.meta.ReadOnlySchemaRepository;
 import com.linkedin.venice.pubsub.PubSubContext;
@@ -14,6 +15,7 @@
 import java.io.IOException;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
+import java.util.function.IntConsumer;
 import org.apache.avro.Schema;
 
 
@@ -34,6 +36,13 @@ public class InternalDaVinciRecordTransformer<K, V, O> extends DaVinciRecordTran
    */
   private final CountDownLatch startLatchConsumptionLatch;
 
+  /**
+   * Wired by {@link com.linkedin.davinci.kafka.consumer.StoreIngestionTask} so that the CDC consumer
+   * can pause/resume Kafka prefetch on its own version's partitions during the AA version swap barrier.
+   */
+  private IntConsumer pausePartitionConsumer;
+  private IntConsumer resumePartitionConsumer;
+
   public InternalDaVinciRecordTransformer(
       DaVinciRecordTransformer recordTransformer,
       Schema keySchema,
@@ -52,6 +61,20 @@ public InternalDaVinciRecordTransformer(
         new CountDownLatch(internalRecordTransformerConfig.getStartConsumptionLatchCount());
   }
 
+  /**
+   * Wires the back-reference from the CDC inner-class transformer to this {@code InternalDaVinciRecordTransformer},
+   * enabling the AA version-swap coordinator to drive Kafka prefetch pause/resume on this version's partitions.
+   * Called from {@link com.linkedin.davinci.kafka.consumer.StoreIngestionTask} after construction; calling this
+   * after construction (rather than inside it) avoids invoking an overridable method on the user's transformer
+   * before initialization is complete.
+   */
+  public void initializeChangelogConsumerBackReference() {
+    if (isCDCRecordTransformer()) {
+      ((VeniceChangelogConsumerDaVinciRecordTransformerImpl.DaVinciRecordTransformerChangelogConsumer) this.recordTransformer)
+          .setInternalRecordTransformer(this);
+    }
+  }
+
   @Override
   public DaVinciRecordTransformerResult<O> transform(
       Lazy<K> key,
@@ -91,13 +114,44 @@ public boolean useUniformInputValueSchema() {
   }
 
   /**
-   * Lifecycle event triggered when a version swap is detected for partitionId
-   * It is used for DVRT CDC.
+   * Lifecycle event triggered when a version swap is detected for partitionId.
+   * It is used for the DaVinciRecordTransformer CDC consumer. The {@code versionSwap} payload is
+   * required for AA-aware version-swap coordination (region filtering, generation-id matching);
+   * it may be null when invoked from legacy or test code paths.
    */
-  public void onVersionSwap(int currentVersion, int futureVersion, int partitionId) {
+  public void onVersionSwap(VersionSwap versionSwap, int currentVersion, int futureVersion, int partitionId) {
     if (isCDCRecordTransformer()) {
       ((VeniceChangelogConsumerDaVinciRecordTransformerImpl.DaVinciRecordTransformerChangelogConsumer) this.recordTransformer)
-          .onVersionSwap(currentVersion, futureVersion, partitionId);
+          .onVersionSwap(versionSwap, currentVersion, futureVersion, partitionId);
+    }
+  }
+
+  /**
+   * Wires the pause/resume handlers used by the AA version-swap coordinator to drive Kafka prefetch
+   * pause/resume on this transformer's own version's partitions.
+   */
+  public void setPartitionPauseHandlers(IntConsumer pauseHandler, IntConsumer resumeHandler) {
+    this.pausePartitionConsumer = pauseHandler;
+    this.resumePartitionConsumer = resumeHandler;
+  }
+
+  /**
+   * Pauses Kafka prefetch for {@code partitionId} on this transformer's version. No-op when handlers
+   * are not wired (e.g. unit tests or non-server contexts).
+   */
+  public void pausePartitionConsumption(int partitionId) {
+    if (pausePartitionConsumer != null) {
+      pausePartitionConsumer.accept(partitionId);
+    }
+  }
+
+  /**
+   * Resumes Kafka prefetch for {@code partitionId} on this transformer's version. No-op when handlers
+   * are not wired.
+   */
+  public void resumePartitionConsumption(int partitionId) {
+    if (resumePartitionConsumer != null) {
+      resumePartitionConsumer.accept(partitionId);
     }
   }