xiangfu0 commented on code in PR #20161:
URL: https://github.com/apache/pinot/pull/20161#discussion_r4477615030
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java:
##########
@@ -58,6 +71,85 @@ public class ConcurrentMapPartitionUpsertMetadataManager
extends BasePartitionUp
public ConcurrentMapPartitionUpsertMetadataManager(String tableNameWithType,
int partitionId, UpsertContext context) {
super(tableNameWithType, partitionId, context);
+ _recoveringLegacySnapshots = _metadataTTL > 0 && !hasVersionedWatermark();
+ _initialSegmentsLoaded = _metadataTTL <= 0;
+ if (_recoveringLegacySnapshots) {
+ _logger.warn("Rebuilding legacy snapshot ownership before trusting the
TTL watermark");
+ }
+ }
+
+ private boolean hasVersionedWatermark() {
+ try {
+ byte[] bytes = Files.readAllBytes(getWatermarkFile().toPath());
+ return bytes.length == 2 * Long.BYTES
+ && ByteBuffer.wrap(bytes).getLong(Double.BYTES) ==
SNAPSHOT_WATERMARK_VERSION;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ @Override
+ protected boolean canSkipSegmentOutOfTTL(double maxComparisonValue) {
+ return !_recoveringLegacySnapshots &&
super.canSkipSegmentOutOfTTL(maxComparisonValue);
+ }
+
+ @Override
+ protected boolean isOutOfMetadataTTL(IndexSegment segment) {
+ // Recovery retains expired keys, so offload must remove them before their
segment is destroyed.
+ return !_recoveringLegacySnapshots && _initialSegmentsLoaded &&
super.isOutOfMetadataTTL(segment);
+ }
+
+ /// Wait for all assigned ONLINE segments, including segments without
snapshots. The first consuming segment
+ /// alone is insufficient for full upsert: it can arrive before the
historical segments.
+ protected boolean isInitialSegmentLoadComplete() {
+ TableDataManager tableDataManager = _context.getTableDataManager();
+ if (tableDataManager == null) {
+ return false;
+ }
+ HelixManager helixManager = tableDataManager.getHelixManager();
+ if (helixManager == null ||
helixManager.getHelixDataAccessor().getProperty(
+
helixManager.getHelixDataAccessor().keyBuilder().idealStates(_tableNameWithType))
== null) {
+ return false;
+ }
+ return TableStateUtils.isAllSegmentsLoaded(helixManager,
_tableNameWithType);
+ }
+
+ @Override
+ protected boolean isReadyToTakeSnapshot() {
+ if (_initialSegmentsLoaded) {
+ return true;
+ }
+ try {
+ if (!isInitialSegmentLoadComplete()) {
+ _logger.info("Defer snapshot and TTL cleanup until all assigned
segments are loaded");
+ return false;
+ }
+ _initialSegmentsLoaded = true;
Review Comment:
**MAJOR:** after a restart, this latch can stay unset indefinitely. Until it
sets, there are no snapshots and no TTL or deleted-key cleanup.
**Why it keeps failing.** The latch is only checked when a consumer starts
(`takeSnapshot` + `removeExpiredPrimaryKeys`). It needs every IS-ONLINE segment
of the table to also be ONLINE in the current state at that moment. On
non-committer replicas, segment N's CONSUMING→ONLINE handler releases the
consumer semaphore before download/build. `ALLOW_ALWAYS` is the default for
full upsert, and under DISALLOW policies the release happens in `doOffload`,
still inside the handler. Helix writes CS=ONLINE only after the handler
returns, so consumer N+1 always sees N as IS=ONLINE/CS=CONSUMING. If the first
check after restart misses (the case this PR targets), every later check misses
too.
**Repro.** Temporary test: real `TableStateUtils` with mocked Helix,
versioned watermark, a restart, then 3 commits.
- This PR: `lastSnapshotPass=null` throughout, keys grow 2→5, persisted
watermark stuck at 290.
- Merge-base: a pass at every consumer start, keys stay at 2, watermark goes
290→400→500.
This hits trusted tables too, not only legacy ones.
**Possible fix.** Latch against the ONLINE set captured at startup (ignore
segments that turn ONLINE later or are mid CONSUMING→ONLINE), or use a periodic
table-level readiness check. Scope the gate to legacy recovery only.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java:
##########
@@ -58,6 +71,85 @@ public class ConcurrentMapPartitionUpsertMetadataManager
extends BasePartitionUp
public ConcurrentMapPartitionUpsertMetadataManager(String tableNameWithType,
int partitionId, UpsertContext context) {
super(tableNameWithType, partitionId, context);
+ _recoveringLegacySnapshots = _metadataTTL > 0 && !hasVersionedWatermark();
+ _initialSegmentsLoaded = _metadataTTL <= 0;
+ if (_recoveringLegacySnapshots) {
+ _logger.warn("Rebuilding legacy snapshot ownership before trusting the
TTL watermark");
+ }
+ }
+
+ private boolean hasVersionedWatermark() {
+ try {
+ byte[] bytes = Files.readAllBytes(getWatermarkFile().toPath());
+ return bytes.length == 2 * Long.BYTES
+ && ByteBuffer.wrap(bytes).getLong(Double.BYTES) ==
SNAPSHOT_WATERMARK_VERSION;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ @Override
+ protected boolean canSkipSegmentOutOfTTL(double maxComparisonValue) {
+ return !_recoveringLegacySnapshots &&
super.canSkipSegmentOutOfTTL(maxComparisonValue);
+ }
+
+ @Override
+ protected boolean isOutOfMetadataTTL(IndexSegment segment) {
+ // Recovery retains expired keys, so offload must remove them before their
segment is destroyed.
+ return !_recoveringLegacySnapshots && _initialSegmentsLoaded &&
super.isOutOfMetadataTTL(segment);
+ }
+
+ /// Wait for all assigned ONLINE segments, including segments without
snapshots. The first consuming segment
+ /// alone is insufficient for full upsert: it can arrive before the
historical segments.
+ protected boolean isInitialSegmentLoadComplete() {
+ TableDataManager tableDataManager = _context.getTableDataManager();
+ if (tableDataManager == null) {
+ return false;
+ }
+ HelixManager helixManager = tableDataManager.getHelixManager();
+ if (helixManager == null ||
helixManager.getHelixDataAccessor().getProperty(
+
helixManager.getHelixDataAccessor().keyBuilder().idealStates(_tableNameWithType))
== null) {
+ return false;
+ }
+ return TableStateUtils.isAllSegmentsLoaded(helixManager,
_tableNameWithType);
+ }
+
+ @Override
+ protected boolean isReadyToTakeSnapshot() {
+ if (_initialSegmentsLoaded) {
+ return true;
+ }
+ try {
+ if (!isInitialSegmentLoadComplete()) {
+ _logger.info("Defer snapshot and TTL cleanup until all assigned
segments are loaded");
+ return false;
+ }
+ _initialSegmentsLoaded = true;
+ return true;
+ } catch (Exception e) {
+ _logger.warn("Cannot verify that initial segment loading is complete,
deferring snapshot and TTL cleanup", e);
+ return false;
+ }
+ }
+
+ @Override
+ protected synchronized void doTakeSnapshot(@Nullable DocIdsSnapshot.Trigger
trigger) {
Review Comment:
Minor: `synchronized` holds the manager monitor for the entire snapshot
pass. Meanwhile, `startOperation()` on Helix threads
(add/replace/remove/preload for this partition) and `stop()` block until the
pass ends, though the pass itself was designed to be non-blocking with
`tryLock`. Snapshots already run serially on the consumer thread, so either
drop the lock or narrow it to the versioned publish.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
