[dvc][cc] Fix unsafe DVRT CDC version swap - #2795
Draft
kvargha wants to merge 16 commits into
Draft
Conversation
Introduce RecordTransformerVersionSwapCoordinator, a barrier that gates
version swap until every assigned partition observes VSMs from every
region on both current and future version topics. Replaces the legacy
per-partition immediate flip, which is incorrect under AA topology
because sibling regions may not have replicated their VSMs yet.
- Coordinator: state machine IDLE -> IN_PROGRESS ->
{COMMITTED,TIMED_OUT,FAILED};
atomic cutover across all assigned partitions on barrier completion.
- Pauses Kafka prefetch on future-side partitions until commit via new
pause/resume hooks on InternalDaVinciRecordTransformer + StoreIngestionTask.
- Watchdog timeout force-commits if the barrier does not close in time.
- Legacy per-partition path preserved when versionSwapByControlMessageEnabled
is false.
Tests: 18 coordinator unit tests, 7 CDC consumer AA tests, and 6 end-to-end
TestAaVersionSwapRecordTransformer scenarios.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Pull request overview
This PR introduces a new RecordTransformerVersionSwapCoordinator that gates the DVRT-based CDC consumer's version-swap cutover until every assigned partition has observed VSMs from every region on both the current and future version topics, then atomically flips partitionToVersionToServe for all partitions in one synchronized block. The coordinator is gated behind the existing versionSwapByControlMessageEnabled flag (default false), so legacy per-partition flip behavior is preserved.
Changes:
- New
RecordTransformerVersionSwapCoordinator(state machine:IDLE → IN_PROGRESS → {COMMITTED, TIMED_OUT, FAILED}) plumbed intoVeniceChangelogConsumerDaVinciRecordTransformerImpl. InternalDaVinciRecordTransformergains pause/resume Kafka-prefetch hooks (wired byStoreIngestionTask) andonVersionSwapnow takes theVersionSwappayload.- New unit + integration tests covering AA cutover, timeout watchdog, rollback, restart, back-to-back swaps, and buffer pressure.
Reviewed changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated 14 comments.
Show a summary per file
| File | Description |
|---|---|
clients/da-vinci-client/.../RecordTransformerVersionSwapCoordinator.java |
New cross-region barrier coordinator with state machine, watchdog, and per-side accumulators |
clients/da-vinci-client/.../VeniceChangelogConsumerDaVinciRecordTransformerImpl.java |
Wires the coordinator, surfaces watchdog failures via poll(), branches between AA and legacy paths in onVersionSwap |
clients/da-vinci-client/.../InternalDaVinciRecordTransformer.java |
New pause/resume handlers and onVersionSwap(VersionSwap, …) signature; back-reference initialization |
clients/da-vinci-client/.../StoreIngestionTask.java |
Wires pause/resume handlers from the SIT into the transformer; passes VersionSwap payload |
clients/da-vinci-client/test/.../RecordTransformerVersionSwapCoordinatorTest.java |
18 unit tests for the coordinator's relevance/accumulation/state-machine behaviour |
clients/da-vinci-client/test/.../VeniceChangelogConsumerDaVinciRecordTransformerAaVersionSwapTest.java |
7 unit tests covering pause/resume, foreign-region/legacy/stale filtering, atomic flip, failure surfacing |
clients/da-vinci-client/test/.../VeniceChangelogConsumerDaVinciRecordTransformerImplTest.java |
Updates legacy onVersionSwap call sites to the new 4-arg signature |
clients/da-vinci-client/test/.../RecordTransformerTest.java |
Updates onVersionSwap call site and adds pause/resume wiring test |
internal/venice-test-common/.../TestAaVersionSwapRecordTransformer.java |
New end-to-end integration tests: pre/post swap, restart, watchdog timeout, back-to-back swaps, rollback, buffer pressure |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
kvargha
changed the title
kvargha
changed the title
worktree-dvrt-cdc-aa-version-swap
Fixes 2 SpotBugs failures in the AA version-swap tests: - DP_DO_INSIDE_DO_PRIVILEGED (5x Field.setAccessible calls in the unit test's setUp + 3 test methods): replaced with @VisibleForTesting getters/setters on VeniceChangelogConsumerDaVinciRecordTransformerImpl. Drops `final` from changeCaptureStats and versionSwapCoordinator so tests can swap them in. Matches the @VisibleForTesting idiom already used 8x in the same production class. - DE_MIGHT_IGNORE (TestAaVersionSwapRecordTransformer.deleteStoreQuietly silently swallowed Exception): now logs at debug level. Verified locally: spotbugsTest + spotbugsIntegrationTest both pass; all 8 tests in VeniceChangelogConsumerDaVinciRecordTransformerAaVersionSwapTest still pass. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
testAaVersionSwapRollback was asserting parent.currentVersion == 2 after rolling back from v3, but a bug in upstream PR linkedin#2785's VeniceParentHelixAdmin.updateParentVersionStatusAfterRollback double-decrements parent.currentVersion when 3+ versions exist: the admin-task handler decrements 3->2 (via VeniceHelixAdmin), then updateParentVersionStatusAfterRollback reads store.getCurrentVersion() (now 2), computes getBackupVersionNumber(versions, 2) = 1, and decrements again to 1. Child controllers decrement correctly to 2, so switch the assertion to getColoToCurrentVersions() which reflects each child's view. This will need to be tracked for upstream fix separately; the test sidesteps the parent-side double decrement. Verified locally: testAaVersionSwapRollback PASSED (222s). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
The class lands in the catch-all IntegrationTests_99 by default. With 7 tests totaling ~13 minutes, that shard overflows the 15-min CI limit (cancelled in PR linkedin#2795 CI). Assign it to its own shard so the runtime sits well within the per-shard budget. This is a manual entry pending the next run of scripts/ci/rebalance_test_shards.py with collected timing data. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
The monolithic TestAaVersionSwapRecordTransformer (~13 min runtime across 7 tests) exceeds the 15-min per-shard CI budget even when given its own shard. Split by topic into two classes that each fit comfortably, and rename to better reflect what's actually under test (the DVRT-based CDC consumer's behavior under AA version swaps, not the RecordTransformer class itself). - AbstractDvrtCdcAaVersionSwapTest — shared multi-region cluster setup, helpers (createAaHybridStore, emptyPush, writeRecords, buildClientConfig, pollUntilRangeObserved, drainForDuration, assertNoLoss, deleteStoreQuietly), and constants. Pattern matches AbstractTestRepush. - TestDvrtCdcAaVersionSwap — single-swap basic + edge scenarios: stateful, stateless, mid-swap restart, watchdog timeout, buffer pressure (~440s). - TestDvrtCdcAaVersionSwapMultiVersion — multi-version scenarios: back-to-back swaps, rollback (~387s). Each subclass lands in its own dedicated shard (86 and 87). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…ivate SpotBugs MS_PKGPROTECT flagged the protected static final String[] — a mutable array exposed to subclasses in any package. Both concrete subclasses are in the same package, so package-private access is sufficient and silences the warning. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Bug fixes: - onVersionSwap AA branch no longer rethrows after failSwap; rethrowing propagated out of StoreIngestionTask.processControlMessage and would have killed per-partition ingestion. The exception is surfaced via the coordinator's failure-surface and observed by the next consumer poll(). - versionSwapThreadException is now cleared after poll() reads it (getAndSet(null)) so the consumer can recover for the next swap instead of throwing forever after one failure. - failSwap always invokes failureSurface even when no swap is armed (state != IN_PROGRESS), so a pre-arm exception (e.g., NPE inside isRelevant on a malformed VSM) is no longer silently swallowed. FAIL metric and the rest of the cleanup remain CAS-guarded. - Test constant TOTAL_REGIONS was overloaded as both region count and partition-count loop bound; split into a separate PARTITION_COUNT. Race fixes: - recordCurrentVsm/recordFutureVsm now re-verify generationId against the active swap after arming, defending against an interleaved arm by a different VSM between the caller's isRelevant check and the record call. - canRecord (new helper) re-checks the max-served-version gate inside the monitor so a swap that committed between isRelevant and record rejects this stale VSM. - canRecord checks subscribedPartitions BEFORE arming, so a VSM for an unsubscribed partition no longer starts the timeout watchdog on a doomed swap whose snapshot wouldn't include the partition anyway. - flipServingVersion now also flips partitions that joined subscribedPartitions after the snapshot was captured. Without this, a late subscriber would remain pinned to the OLD version with future-version records silently filtered until the next swap. Defense-in-depth: - Null guards on VSM.oldServingVersionTopic, newServingVersionTopic, and destinationRegion in both isRelevant and canRecord. - handleTerminalFailure now resumes both sides' paused partitions so SIT doesn't continue running with prefetch stuck paused after a flip failure. - flipServingVersion bails out (with an error log) if activeNewVersion hasn't been set, instead of poisoning every assigned partition with -1. - shutdownNow() is called outside the coordinator's monitor so it can't compound the wait if the watchdog task is mid-timeoutSwap. - armIfNeeded only proceeds from IDLE (not just "not IN_PROGRESS"), giving a tighter invariant that survives changes to terminal-state ordering. - Constructor rejects non-positive versionSwapTimeoutInMs. Cleanup: - State field is plain (not AtomicReference) — every access is already inside synchronized methods. - poll()'s VeniceException message no longer has a trailing colon before the wrapped cause. - Test files use proper imports instead of fully-qualified type names. New tests cover: pre-arm failure surfaces and is cleared by next poll; late-subscriber flip; null VSM fields rejected; terminal-failure resumes both sides; unsubscribed-partition VSM doesn't arm; constructor validates timeout.
… IT base - Plumb LogContext through the coordinator's constructor and on into the watchdog DaemonThreadFactory so the timeout-thread log lines carry the CDC component identity (matches the rest of the consumer's thread factories). - Make AbstractDvrtCdcAaVersionSwapTest#deleteStoreQuietly synchronous. The previous fire-and-forget CompletableFuture.runAsync could race the next test's setup on the controller; running cleanup inline on the caller's @AfterMethod / finally thread eliminates that without changing the best-effort error-handling contract.
The lines-added check (Enforce Max Lines Added Per File workflow) caps each production file at 500 added lines vs main. After the review-fix pass the coordinator was at 605; this commit drops it to 495 via: - Drop the convenience (no-LogContext) constructor; require LogContext at construction. Tests pass `(LogContext) null` where they don't need one. - Consolidate the near-identical recordCurrentVsm and recordFutureVsm into a shared private recordVsm helper; the public methods are now 3 lines each. - Likewise extract resumeSide(transformer, paused) helper used by both resumeCurrentSide and resumeFutureSide. - Inline computeMaxServedVersion as a stream expression. - Tighten verbose javadoc and inline comments (no semantics changed).
Two correctness fixes surfaced by self-review: - The consumer's failure surface was `versionSwapThreadException::set`, which unconditionally overwrites any prior exception. If two swap failures land before the user polls, the first is silently dropped. Switch to a new stashVersionSwapException helper that chains via accumulateAndGet + Throwable#addSuppressed so both failures surface from the next poll(). - commitSwap rethrew after handleTerminalFailure had already run all the cleanup (FAIL metric, surface, resumes, state clear). The production caller in onVersionSwap then re-entered failSwap (which became a no-op via state check) — but the rethrow itself was misleading and made tests expect a propagation that production swallows. Now commitSwap absorbs the flip failure entirely. Updated the coordinator unit test to match. Also dropped the unreachable `activeNewVersion <= 0` guard in flipServingVersion; armIfNeeded always sets it before transitioning to IN_PROGRESS, and CAS guards ensure flipServingVersion is only called from IN_PROGRESS. The guard's existence was dead code with no test coverage. New unit test (testOnVersionSwapAaPathAccumulatesConcurrentFailures) verifies the first failure remains the primary cause and the second is attached as a suppressed exception, then poll() surfaces both.
Addresses Copilot review: a VSM whose newServingVersionTopic parses to 0 (Store.NON_EXISTING_VERSION) would slip past the max-served-version gate because computeMaxServedVersion() returns -1 when partitionToVersionToServe is empty (0 > -1). Added an explicit newVersion > Store.NON_EXISTING_VERSION check in both isRelevant and canRecord so a malformed VSM targeting v0 can't arm a swap. New unit test testIsRelevantRejectsNonExistingTargetVersion covers both isRelevant and recordCurrentVsm paths.
…branches Addresses Copilot review on quadratic scan cost: every isRelevant() / canRecord() invocation was scanning partitionToVersionToServe to compute the max served version, compounding to O(partitions²) per swap on large stores. Take the snapshot once in armIfNeeded and serve it back from computeMaxServedVersion() while state == IN_PROGRESS. partitionToVersionToServe only changes mid-swap via the coordinator's own flipServingVersion (called once at commit), so the snapshot is consistent for the duration of the swap. Off-arm calls (rare — only when a new swap is being armed) still scan. Also collapsed the isRelevant() if/else block selecting old vs new topic into a single ternary check to keep the file under the 500-line limit.
The post-swap settle drain was 15s — well past the 5s watchdog + ~2s VSM-detection cycle, leaving 8s of idle wait per swap-triggering test. Trim it to 8s (1s detection + 5s watchdog + 2s controller-jitter buffer) and drop the consumer's VSM detection interval from 3s -> 1s so the happy path observes both regions' VSMs sooner. Net savings per swap: ~9s. With 17 swaps across the suite, that's ~2.5 minutes shaved off the combined IntegrationTests_86/87 wall-clock. No coverage lost — the same number of swaps, same assertions, same edge cases. The watchdog test still exercises the timeout path: the fake "third region" never appears so the watchdog still has to fire to force the cutover.
Comment on lines
+374
to
+388
| public synchronized void handleUnsubscribe(Set<Integer> partitions) { | ||
| for (int partition: partitions) { | ||
| assignedPartitionsSnapshot.remove(partition); | ||
| currentVersionRegionsConsumed.remove(partition); | ||
| futureVersionRegionsConsumed.remove(partition); | ||
| if (pausedCurrentPartitions.remove(partition) && currentTransformerRef != null) { | ||
| currentTransformerRef.resumePartitionConsumption(partition); | ||
| } | ||
| if (pausedFuturePartitions.remove(partition) && futureTransformerRef != null) { | ||
| futureTransformerRef.resumePartitionConsumption(partition); | ||
| } | ||
| } | ||
| if (state == State.IN_PROGRESS && !assignedPartitionsSnapshot.isEmpty() && allPartitionsBothSidesComplete()) { | ||
| commitSwap(); | ||
| } |
Comment on lines
+270
to
+292
| public synchronized void commitSwap() { | ||
| if (state != State.IN_PROGRESS) { | ||
| return; | ||
| } | ||
| state = State.COMMITTED; | ||
| cancelWatchdog(); | ||
| try { | ||
| flipServingVersion(); | ||
| resumeFutureSide(); | ||
| } catch (Exception flipFailure) { | ||
| handleTerminalFailure(flipFailure); | ||
| return; | ||
| } | ||
| if (changeCaptureStats != null) { | ||
| changeCaptureStats.emitVersionSwapCountMetrics(SUCCESS); | ||
| } | ||
| LOGGER.info( | ||
| "Version swap committed for store: {}, swap: {} -> {}, partitions: {}", | ||
| storeName, | ||
| activeOldVersionTopic, | ||
| activeNewVersionTopic, | ||
| assignedPartitionsSnapshot); | ||
| clearSwapState(); |
Comment on lines
+218
to
+219
| (isCurrentSide ? pausedCurrentPartitions : pausedFuturePartitions).add(partition); | ||
| return true; |
…ntime" This reverts 4ad082e. Empirical results from the run on that commit: Shard 86 (TestDvrtCdcAaVersionSwap): before: 633s after: 618s (saved 15s) Shard 87 (TestDvrtCdcAaVersionSwapMultiVersion): before: BackToBack 165.7s | Rollback 134.1s after: BackToBack 170.5s | Rollback 206.8s FAILED + 171.4s retry pass The rollback test got ~37s slower per pass AND failed on the first attempt ("Key 16 not yet observed" — the post-rollback v4 swap didn't deliver the new records within the 90s poll window). Either the shorter drain or the tighter Helix detection interval starves the rollback flow; isolating which isn't worth the marginal 15s win on shard 86. Restoring the previous timing constants — the suite is back to a stable ~633s/466s baseline.
|
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. |
Captures the problem, coordinator design, file map, and remaining CI/review work for PR linkedin#2795 so the context survives across sessions. Planning artifact; to be removed before merge. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Comment on lines
+386
to
+388
| if (state == State.IN_PROGRESS && !assignedPartitionsSnapshot.isEmpty() && allPartitionsBothSidesComplete()) { | ||
| commitSwap(); | ||
| } |
Comment on lines
+437
to
+451
| private void flipServingVersion() { | ||
| // armIfNeeded sets activeNewVersion > 0 before state transitions to IN_PROGRESS, and | ||
| // commitSwap/timeoutSwap only call this from IN_PROGRESS via state guard. So activeNewVersion | ||
| // is always > 0 here by construction. | ||
| for (int partition: assignedPartitionsSnapshot) { | ||
| partitionToVersionToServe.put(partition, activeNewVersion); | ||
| } | ||
| // Late subscribers (joined after the snapshot) get flipped too so future-version records aren't | ||
| // silently filtered until the next swap. | ||
| for (int partition: subscribedPartitions) { | ||
| if (!assignedPartitionsSnapshot.contains(partition)) { | ||
| partitionToVersionToServe.put(partition, activeNewVersion); | ||
| } | ||
| } | ||
| } |
Comment on lines
+3
to
+13
| 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. |
kvargha
marked this pull request as draft
June 26, 2026 21:28
|
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. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem Statement
This is the DVRT-based CDC counterpart to #2280, which added the same version-swap-by-control-message handling to the legacy
VeniceChangelogConsumerImpl. Both PRs sit on top of #2245, which introduced the multi-region VSM broadcast on the controller/server side.The current version-swap behavior in
VeniceChangelogConsumerDaVinciRecordTransformerImplis unsafe in three ways:sourceRegion), but today's per-partition flip ignoressourceRegionand acts on the first VSM it sees.generationIdandoldServingVersionTopic, the consumer can be misled into a swap by a stale message left over from an earlier push (or by historical VSMs replayed after restarting from EARLIEST, or after a rollback to a lower version).poll()before more are ingested), but the future-version transformer's ingestion bypasses user post-poll entirely. Whenever post-poll is slow — or under A/A where DCs ingest at different rates across versions — the future version's transformer can consume past the swap point before the current version has reached it. Records that should have surfaced on the current side are then silently swallowed by the future side: data loss at swap time.Solution
Introduce
RecordTransformerVersionSwapCoordinatorand route the DVRT CDC consumer'sonVersionSwapthrough it whenversionSwapByControlMessageEnabled = true. The coordinator addresses the three issues above:isRelevant()drops VSMs whosesourceRegionis not this client's region.isRelevant()drops VSMs withgenerationId == -1, VSMs targeting a version not greater than the highest already promoted to serving (defends against rollback + replay-from-EARLIEST), and — once a swap is armed — VSMs whosegenerationIdoroldServingVersionTopic/newServingVersionTopicdon't match the in-progress swap.destinationRegionon the future side, the coordinator pauses Kafka prefetch on that partition until the cross-partition barrier closes. The cutover atomically flipspartitionToVersionToServeand resumes the paused partitions; a watchdog force-commits if the barrier doesn't close withinversionSwapTimeoutInMs(default 30 min), bounding the vulnerable window.Backward compatible: legacy per-partition flip is preserved when the config is
false(the default).Trade-off (documented in the coordinator's class javadoc): Kafka's
consumer.pause()doesn't truncate the in-flight batch, so records past the VSM in the same batch flow throughprocessPuton both sides — surfaced on current, silently dropped on future. The design accepts this assuming approximate poll-batch alignment between current and future leaders; the watchdog bounds the window.Code changes
versionSwapByControlMessageEnabled(defaultfalse). Watchdog uses existingversionSwapTimeoutInMs(default 30 min). No new configs.Concurrency-Specific Checks
synchronizedon the instance.poll(), not silently dropped.How was this PR tested?
RecordTransformerVersionSwapCoordinatorTest(18),VeniceChangelogConsumerDaVinciRecordTransformerAaVersionSwapTest(7).TestAaVersionSwapRecordTransformer(6 e2e scenarios — pre/post-swap, mid-swap restart, watchdog timeout, back-to-back swaps, rollback, buffer pressure).onVersionSwap()signature.falsekeeps legacy behavior.Does this PR introduce any user-facing or breaking changes?
versionSwapByControlMessageEnabled(defaultfalse).