xiangfu0 commented on code in PR #19735:
URL: https://github.com/apache/pinot/pull/19735#discussion_r4290480550
##########
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:
This writes IdealState directly through the dedicated multi-write ZkClient,
which deliberately bypasses Helix's `IDEAL_STATE_GROUP_COMMIT` that the normal
delete path (`removeSegmentsFromIdealState`) uses. The version CAS keeps it
correct, but under a busy table it competes with Helix's own IdealState writers
via optimistic retries rather than being batched, so cleanup can exhaust the
retries and skip (fail-safe). Could you add a short comment here noting the
group-commit bypass is intentional and required so the IdealState removal can
share the lineage transaction?
--
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]
