Defer TTL watermark bump until after segment add/preload/replace (upsert & dedup) - #19512
Defer TTL watermark bump until after segment add/preload/replace (upsert & dedup)#19512deepthi912 wants to merge 1 commit into
Conversation
2dc6e78 to
2e08c3c
Compare
deepthi912
changed the title
2e08c3c to
0367b72
Compare
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19512 +/- ##
============================================
- Coverage 67.73% 67.68% -0.06%
Complexity 1430 1430
============================================
Files 3489 3490 +1
Lines 224637 224952 +315
Branches 35468 35517 +49
============================================
+ Hits 152160 152259 +99
- Misses 60454 60657 +203
- Partials 12023 12036 +13
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
KKcorps
left a comment
KKcorps
left a comment
There was a problem hiding this comment.
Non-blocking follow-up: ConsumerCoordinator ordering during rebalance
There is a separate, pre-existing ordering gap worth tracking after this watermark change. With enforceConsumptionInOrder=true, the coordinator tracks the maximum registered sequence; it does not check that every earlier required segment has finished loading.
A concrete FULL-upsert case on an already-ready rebalance target:
- S0 (seq 0) is ONLINE with K@100 and watermark 100. A (seq 1, K@190) starts loading and pauses before row insertion.
- B (seq 2, unrelated J@200) finishes loading, publishes watermark 200, and registers seq 2.
- C (seq 3), the only consumer, passes the predecessor check because seq 2 is registered. With preload disabled and no consuming row processed yet, its first FULL-upsert snapshot is skipped; startup TTL cleanup at cutoff 170 removes K's old metadata pointer while A is still loading.
- A resumes and inserts K@190 as a new key, leaving both S0 and A valid.
Helix serializes transitions per segment, so the PR description's serialization claim should distinguish a Helix partition (segment) from the shared stream partition.
I reproduced the metadata interleaving and checked the coordinator path against source; I have not run an end-to-end Helix rebalance or measured its frequency in production. I have not established another production path for this overlap finding.
Let's keep this non-blocking for this PR and address the ordering guarantee in ConsumerCoordinator and its segment-registration/readiness integration as follow-up work: DATA-3415.
8d9d26f to
eb47d55
Compare
…record add (upsert & dedup) Segment-level: `BasePartitionUpsertMetadataManager.doAddSegment`, `doPreloadSegment`, `doReplaceSegment` and `BasePartitionDedupMetadataManager.addSegment`, `preloadSegment`, `replaceSegment` advanced the TTL watermark (`_largestSeenComparisonValue` / `_largestSeenTime`) up-front, before the segment's rows were inserted into the primary-key map. A concurrent `removeExpiredPrimaryKeys` sweep on another thread reads the newly- advanced watermark and can expire pointers for primary keys the in-flight add is about to insert, producing duplicate first-row inserts on those keys. Per-record: `ConcurrentMapPartitionUpsertMetadataManager.doAddRecord`, `ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes.doAddRecord`, and `ConcurrentMapPartitionDedupMetadataManager.checkRecordPresentOrUpdate` have the same shape at record granularity: they bump the watermark before the `compute()` call that installs or updates the record. In the narrow window between the bump and the compute, the sweep can expire the existing entry for the key being updated; the compute then takes the first-write branch instead of the replace branch, leaving both the old-segment row (still valid-bit-set) and the new-segment row query-visible. Move all bumps to after the row/record is installed. For dedup, this also means the compute lambda's `isOutOfMetadataTTL` check evaluates staleness against the historical state, which is the intended dedup semantic. For dedup base class, drop the `updateWatermark` parameter from `skipSegmentOutOfTTL` and introduce `updateLargestSeenTime` as the mutation helper called at the correct point. If the add throws, the watermark stays unchanged; bumping on failure would let the sweep expire other keys against a phantom watermark justified by a segment we did not actually process. Helix retries the failed transition and re-bumps legitimately on success. Per-partition state transitions are serialized by Helix per segment, so concurrent add invocations for the same segment are not possible; the only concurrency here is with the sweep thread and with other segments' loads. A separate cross-segment race (one segment's bump enabling sweep expiry on keys another concurrent segment is still loading) is tracked as follow-up DATA-3415. Existing 72 upsert and dedup unit tests pass unchanged.
eb47d55 to
ae9e900
Compare
Jackie-Jiang
left a comment
Jackie-Jiang
left a comment
There was a problem hiding this comment.
Trying to understand why this race can even happen. When this race happens, it means the records within the same segment span over TTL. This itself is invalid, and undefined. I don't think we can fix it in any way. Imagine a segment is being added, in the meanwhile a consuming record pushes the watermark higher, and we run into the same problem.
Also, why is an uploaded segment moving the watermark? Shouldn't it always be moved by consuming segment?
| if (skipSegmentOutOfTTL(segment)) { | ||
| updateLargestSeenTime(segment); | ||
| } else { | ||
| addOrReplaceSegment(null, segment); | ||
| updateLargestSeenTime(segment); | ||
| } |
There was a problem hiding this comment.
(minor) Can be simplified, same for other places
| if (skipSegmentOutOfTTL(segment)) { | |
| updateLargestSeenTime(segment); | |
| } else { | |
| addOrReplaceSegment(null, segment); | |
| updateLargestSeenTime(segment); | |
| } | |
| if (!skipSegmentOutOfTTL(segment)) { | |
| addOrReplaceSegment(null, segment); | |
| } | |
| updateLargestSeenTime(segment); |
| updateLargestSeenTime(segment); | ||
| return; | ||
| } | ||
| try (DedupUtils.DedupRecordInfoReader dedupRecordInfoReader = new DedupUtils.DedupRecordInfoReader(segment, |
There was a problem hiding this comment.
(minor) Remove the previous check and put this in if(!skipSegmentOutOfTTL(segment)) for better readability
4 participants
Problem
Both upsert (
BasePartitionUpsertMetadataManager.doAddSegment/doPreloadSegment/doReplaceSegment) and dedup (BasePartitionDedupMetadataManager.addSegment/preloadSegment/replaceSegment) advance the TTL watermark (_largestSeenComparisonValue/_largestSeenTime) before the segment's rows are inserted into the primary-key map. A concurrentremoveExpiredPrimaryKeyssweep on another thread reads the newly-advanced watermark and can expire pointers for primary keys the in-flight add is about to insert, producing duplicate first-row inserts on those keys.Race timeline (upsert
doAddSegmentfor segment withmax=1000,metadataTTL=300, previous watermark500):_largestSeenComparisonValuefrom 500 → 1000.removeExpiredPrimaryKeys) reads watermark = 1000, expires all PKs with comparison value< 700.Dedup has the same race in the segment-level pre-bump inside
skipSegmentOutOfTTL(segment, updateWatermark=true).Fix
Move the watermark bump to after the add/preload/replace call returns. The out-of-TTL skip path bumps the watermark before returning as before, since it inserts no rows and cannot race with the sweep.
For dedup, drop the
updateWatermarkparameter onskipSegmentOutOfTTL(was alwaystruefor adds andfalsefor removes) and introduce a smallupdateLargestSeenTimehelper that callers invoke at the correct point.If the add throws, the watermark stays unchanged. Bumping on failure would let the sweep expire other keys against a phantom watermark justified by a segment we did not actually process; Helix retries the failed transition and the retry re-computes and bumps legitimately.
Per-partition state transitions are serialized by Helix, so concurrent add invocations on the same partition are not possible; the only concurrency here is with the sweep thread.