xiangfu0 commented on code in PR #19735:
URL: https://github.com/apache/pinot/pull/19735#discussion_r4290482703
##########
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());
+ }
+ builder.execute();
+
+ TableConfig tableConfig =
ZKMetadataProvider.getTableConfig(_propertyStore, tableNameWithType);
+ _segmentDeletionManager.deleteSegments(tableNameWithType,
segmentNames, tableConfig);
Review Comment:
The physical segment delete runs only after the transaction commits, so a
controller crash in this window leaves segments out of IdealState with their
znodes still present. That matches the old non-atomic ordering so it's not a
regression, and the start/revert paths keep the lineage entry around so a later
retention pass reaps the orphans. In the RetentionManager all-orphan branch the
entry is already removed, so those znodes can become permanent orphans (also
pre-existing). Just flagging it; no change needed here.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4566,7 +4656,12 @@ public String startReplaceSegments(String
tableNameWithType, List<String> segmen
// is safe to physically delete segments.
if (!segmentsToCleanUp.isEmpty()) {
LOGGER.info("Cleaning up the segments while startReplaceSegments: {}",
segmentsToCleanUp);
- deleteSegmentsForLineageCleanup(tableNameWithType, segmentsToCleanUp);
+ PinotResourceManagerResponse cleanupResponse =
+ deleteSegmentsForLineageCleanup(tableNameWithType,
segmentsToCleanUp, cleanupLineageVersion[0]);
Review Comment:
This relies on an implicit invariant: cleanupLineageVersion[0] stays -1
unless a lineage write succeeded, and segmentsToCleanUp is only populated in
that same success branch, so a non-empty batch always carries a real version.
That's correct today but easy to break in a later edit. A one-line comment
stating 'non-empty batch implies the version was set on a successful write'
would harden it.
--
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]
