xiangfu0 commented on code in PR #19735:
URL: https://github.com/apache/pinot/pull/19735#discussion_r4278284736
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4882,11 +5020,11 @@ private boolean writeLineageEntryWithLock(TableConfig
tableConfig, String lineag
// Write back to the lineage entry
if (SegmentLineageAccessHelper.writeSegmentLineage(propertyStore,
segmentLineageToUpdate, expectedVersion)) {
- return true;
+ return expectedVersion + 1;
Review Comment:
Corroborating @Jackie-Jiang's P1: the new version bump fences the lineage
*write*, but this retry loop only re-fetches the record and re-checks
`currentLineageEntry.equals(lineageEntryToMatch)` (entry contents), not the
source-segment-ONLINE validation done earlier in
`endReplaceSegments`/`revertReplaceSegments`. So if retention cleanup deletes
source `A` and bumps the version without touching this entry, the retry
re-fetches the newer version, the entry still matches, and the write commits —
routing goes back to the now-deleted `A`, and revert can then also delete `B`.
The outer lifecycle check needs to be bound to the validated version and re-run
on a version change, not just the write.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/lineage/DefaultLineageManager.java:
##########
@@ -107,31 +122,38 @@ public void updateLineageForRetention(TableConfig
tableConfig, SegmentLineage li
segmentsToDelete.addAll(sourceSegments);
}
}
- } else if (lineageEntry.getState() == LineageEntryState.REVERTED || (
- lineageEntry.getState() == LineageEntryState.IN_PROGRESS &&
lineageEntry.getTimestamp()
- < System.currentTimeMillis() - lineageCleanupRetentionMs)) {
+ } else if (isZombieEntryEligibleForCleanup(lineageEntry,
lineageCleanupThresholdMs)) {
Review Comment:
Agree with @Jackie-Jiang's P1 here. For a recent `REVERTED A->B` plus
`COMPLETED A->B`, the reverted entry is retained, so
`filterSegmentsBasedOnLineageInPlace` in the broker preselector excludes `B`
for the reverted entry and `A` for the completed entry — neither is routable
even though `B` survives on disk. The protection is also fragile across passes:
once `A` is retention-deleted the completed entry is pruned, and when the
reverted entry ages past the window `B` becomes an unprotected delete
candidate. Protecting the shared destination should be decoupled from the
zombie grace period, and a multi-pass test should assert broker routing in
addition to physical preservation.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java:
##########
@@ -543,9 +553,15 @@ private void
manageSegmentLineageCleanupForTable(TableConfig tableConfig) {
}
// Delete segments based on the segment lineage
if (!segmentsToDelete.isEmpty()) {
-
_pinotHelixResourceManager.deleteSegmentsForLineageCleanup(tableNameWithType,
segmentsToDelete);
- LOGGER.info("Finished cleaning up segment lineage for table: {} in {}ms,
deleted segments: {}",
- tableNameWithType, (System.currentTimeMillis() - cleanupStartTime),
segmentsToDelete);
+ PinotResourceManagerResponse cleanupResponse =
_pinotHelixResourceManager.deleteSegmentsForLineageCleanup(
+ tableNameWithType, segmentsToDelete, cleanupLineageVersion[0]);
+ if (cleanupResponse.isSuccessful()) {
+ LOGGER.info("Finished cleaning up segment lineage for table: {} in
{}ms, deleted segments: {}",
+ tableNameWithType, (System.currentTimeMillis() -
cleanupStartTime), segmentsToDelete);
+ } else {
+ LOGGER.warn("Skipped segment deletion after lineage cleanup for table:
{}. {}", tableNameWithType,
Review Comment:
+1 on @Jackie-Jiang's P2. When the version fence rejects deletion this
branch only logs, but `updateLineageForRetention` has already removed the
zombie entry and committed it — so an orphan destination (absent from
IdealState, znode still present) can't be rediscovered on a later pass, and
OFFLINE REFRESH tables skip ordinary retention so it can leak indefinitely.
Worth keeping the cleanup intent durable until the delete actually succeeds.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1136,29 +1138,89 @@ public PinotResourceManagerResponse
deleteSegments(String tableNameWithType, Lis
/// @return Request response
public PinotResourceManagerResponse deleteSegments(String tableNameWithType,
List<String> segmentNames,
@Nullable String retentionPeriod) {
- return deleteSegmentsInternal(tableNameWithType, segmentNames,
retentionPeriod, false);
+ return deleteSegmentsInternal(tableNameWithType, segmentNames,
retentionPeriod);
}
- /// Lineage-aware delete path that skips the cross-check against the live
lineage entries. Reserved for callers
- /// that have already coordinated with the lineage lifecycle: proactive
cleanup in `startReplaceSegments`,
- /// post-revert cleanup in `revertReplaceSegments`, and `RetentionManager`'s
lineage-cleanup pass.
+ /// Lineage-aware delete path for callers that have already derived an
authorized cleanup batch from a specific
+ /// lineage version. The lineage znode is version-bumped in the same
ZooKeeper transaction that removes the
+ /// segments from IdealState. This fences every concurrent lineage mutation:
either it commits first and this
+ /// delete is rejected, or this delete commits first and the mutation must
retry against the updated lineage and
+ /// IdealState.
+ ///
/// External call sites (REST handlers, retention based on table config,
minion task generators, push-failure
/// cleanup) must continue to use the public [#deleteSegments] overloads.
+ ///
+ /// @param expectedLineageVersion lineage version used to derive
`segmentNames`
public PinotResourceManagerResponse deleteSegmentsForLineageCleanup(String
tableNameWithType,
- List<String> segmentNames) {
- return deleteSegmentsInternal(tableNameWithType, segmentNames, null, true);
+ List<String> segmentNames, int expectedLineageVersion) {
+ if (segmentNames.isEmpty()) {
+ return PinotResourceManagerResponse.success("No segments to delete");
+ }
+ try {
+
Preconditions.checkArgument(TableNameBuilder.isTableResource(tableNameWithType),
+ "Table name: %s is not a valid table name with type suffix",
tableNameWithType);
+
+ String lineagePath =
ZKMetadataProvider.constructPropertyStorePathForSegmentLineage(tableNameWithType);
+ String fullLineagePath =
PropertyPathBuilder.propertyStore(_helixClusterName) + lineagePath;
+ PropertyKey idealStateKey =
_helixDataAccessor.keyBuilder().idealStates(tableNameWithType);
+
+ for (int attempt = 0; attempt <
DEFAULT_SEGMENT_LINEAGE_UPDATE_NUM_RETRY; attempt++) {
+ ZNRecord lineageRecord =
+
SegmentLineageAccessHelper.getSegmentLineageZNRecord(_propertyStore,
tableNameWithType);
+ if (lineageRecord == null || lineageRecord.getVersion() !=
expectedLineageVersion) {
+ return PinotResourceManagerResponse.failure(
+ "Segment lineage changed before cleanup for table: " +
tableNameWithType);
+ }
+
+ IdealState idealState = _helixDataAccessor.getProperty(idealStateKey);
+ if (idealState == null) {
+ return PinotResourceManagerResponse.failure("IdealState does not
exist for table: " + tableNameWithType);
+ }
+ IdealState updatedIdealState = HelixHelper.cloneIdealState(idealState);
+ boolean idealStateChanged =
updatedIdealState.getPartitionSet().removeAll(segmentNames);
+
+ try {
+ // This transaction deliberately spans the property-store and
IdealState roots, so both paths are absolute.
+ ZkMultiWriteBuilder builder = new
ZkMultiWriteBuilder(getOrBuildMultiWriteZkClient(), "")
+ .set(fullLineagePath, lineageRecord, expectedLineageVersion);
+ if (idealStateChanged) {
+ builder.set(idealStateKey.getPath(),
updatedIdealState.getRecord(), idealState.getRecord().getVersion());
Review Comment:
Confirm-only (non-blocking): this writes the IdealState znode directly via a
raw cross-root `multi()` using the dedicated `ZNRecordSerializer` client,
bypassing `HelixHelper.removeSegmentsFromIdealState`/the Helix data accessor.
Functionally it matches (same partition-set removal, CAS on the accessor's
znode version) and a ZK `multi()` across IDEALSTATES + PROPERTYSTORE roots is
valid. Just worth double-checking the serializer/version read via
`_helixDataAccessor.getProperty(...)` stays byte-compatible with Helix's own IS
writer across versions, since this becomes a second writer to that znode.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4407,11 +4469,28 @@ public String startReplaceSegments(String
tableNameWithType, List<String> segmen
segment, tableNameWithType);
}
List<String> segmentsToCleanUp = new ArrayList<>();
+ int[] cleanupLineageVersion = {-1};
int attemptCount;
synchronized (getLineageUpdaterLock(tableNameWithType)) {
try {
attemptCount = DEFAULT_RETRY_POLICY.attempt(() -> {
long startReplaceSegmentsTsForAttempt = System.currentTimeMillis();
+ Set<String> segmentsToCleanUpForAttempt = new LinkedHashSet<>();
+ Map<String, Integer> authorizedCleanupReferenceCounts = new
HashMap<>();
+
+ // Revalidate against the current IdealState on every
optimistic-write attempt. A version-fenced lineage
+ // cleanup bumps the lineage version in the same transaction that
removes segments from IdealState, so a
+ // concurrent cleanup either makes this validation fail or makes the
lineage write retry.
+ Set<String> segmentsForTableForAttempt = new
HashSet<>(getSegmentsFor(tableNameWithType, true));
Review Comment:
Minor/behavioral: these per-attempt `checkState` re-validations run inside
the `DEFAULT_RETRY_POLICY.attempt` lambda, where a thrown exception is *not*
retried (it propagates as `RetriableOperationException`). So a transient race
that momentarily removes a `segmentsFrom` segment now aborts
`startReplaceSegments` outright instead of retrying. That's fail-closed and
probably fine, but it's a behavior change from the single pre-lock check —
worth noting in the method doc.
--
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]
