suvodeep-pyne commented on code in PR #19995:
URL: https://github.com/apache/pinot/pull/19995#discussion_r4344943584
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2278,6 +2283,80 @@ private ZkClient getOrBuildMultiWriteZkClient() {
}
}
+ /// Removes the ZK metadata of the given segments that are IN_PROGRESS and
not in the IdealState of the table, i.e.
+ /// new consuming segments that were not added to the IdealState, and
returns the segments whose ZK metadata was
+ /// removed.
+ ///
+ /// The removal is submitted in one ZK `multi()` transaction with
version-checked deletes of the segment ZK metadata
+ /// read here, and a version-checked write of the unchanged IdealState:
+ /// - An IdealState update that read the IdealState before the transaction
fails its version check. An updater that
+ /// reads the segment ZK metadata again when it retries, like the one of
RealtimeSegmentValidationManager, then no
+ /// longer finds the removed segments.
+ /// - If the IdealState, or the ZK metadata of one of the segments, is
updated after it was read here, the transaction
+ /// fails and is retried.
+ ///
+ /// The data versions cannot tell a segment ZK metadata deleted and
recreated with the same name since it was read,
+ /// e.g. by an upload under the same name, from the one that was read. As
the IdealState is outside the property
+ /// store, the transaction uses the multi-write ZkClient directly instead of
[ZkMultiWriteBuilder].
+ public Set<String> removeInProgressSegmentsZKMetadataNotInIdealState(String
tableNameWithType,
+ Collection<String> segmentNames) {
+ return
removeInProgressSegmentsZKMetadataNotInIdealState(tableNameWithType,
segmentNames, () -> {
+ });
+ }
+
+ /// @param afterRead Runs after reading the IdealState and the segment ZK
metadata, before removing it
+ @VisibleForTesting
+ Set<String> removeInProgressSegmentsZKMetadataNotInIdealState(String
tableNameWithType,
+ Collection<String> segmentNames, Runnable afterRead) {
+ ZkClient zkClient = getOrBuildMultiWriteZkClient();
+ String idealStatePath = PropertyPathBuilder.idealState(_helixClusterName,
tableNameWithType);
+ String propertyStoreRoot =
PropertyPathBuilder.propertyStore(_helixClusterName);
+ for (int attempt = 1; ; attempt++) {
+ Stat idealStateStat = new Stat();
+ ZNRecord idealState = zkClient.readData(idealStatePath, idealStateStat);
+ List<Op> segmentDeletes = new ArrayList<>();
+ Set<String> segmentsToRemove = new HashSet<>();
+ for (String segmentName : segmentNames) {
+ if (idealState.getMapFields().containsKey(segmentName)) {
+ continue;
+ }
+ String segmentPath =
+ propertyStoreRoot +
ZKMetadataProvider.constructPropertyStorePathForSegment(tableNameWithType,
segmentName);
+ Stat segmentStat = new Stat();
+ ZNRecord segmentRecord;
+ try {
+ segmentRecord = zkClient.readData(segmentPath, segmentStat);
+ } catch (ZkNoNodeException e) {
+ continue;
+ }
+ if (new SegmentZKMetadata(segmentRecord).getStatus() ==
CommonConstants.Segment.Realtime.Status.IN_PROGRESS) {
+ segmentDeletes.add(Op.delete(segmentPath, segmentStat.getVersion()));
+ segmentsToRemove.add(segmentName);
+ }
+ }
+ if (segmentsToRemove.isEmpty()) {
+ return segmentsToRemove;
+ }
+ afterRead.run();
+ List<Op> ops = new ArrayList<>(segmentDeletes.size() + 1);
+ ops.add(Op.setData(idealStatePath, zkClient.serialize(idealState,
idealStatePath), idealStateStat.getVersion()));
+ ops.addAll(segmentDeletes);
+ try {
+ zkClient.multi(ops);
+ return segmentsToRemove;
+ } catch (ZkException e) {
Review Comment:
Added a comment in df8842da90. ZkMultiWriteBuilder can't take the IdealState
path, since it only writes property-store paths, so this call site unwraps the
same way inline.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2278,6 +2283,80 @@ private ZkClient getOrBuildMultiWriteZkClient() {
}
}
+ /// Removes the ZK metadata of the given segments that are IN_PROGRESS and
not in the IdealState of the table, i.e.
+ /// new consuming segments that were not added to the IdealState, and
returns the segments whose ZK metadata was
+ /// removed.
+ ///
+ /// The removal is submitted in one ZK `multi()` transaction with
version-checked deletes of the segment ZK metadata
+ /// read here, and a version-checked write of the unchanged IdealState:
+ /// - An IdealState update that read the IdealState before the transaction
fails its version check. An updater that
+ /// reads the segment ZK metadata again when it retries, like the one of
RealtimeSegmentValidationManager, then no
+ /// longer finds the removed segments.
+ /// - If the IdealState, or the ZK metadata of one of the segments, is
updated after it was read here, the transaction
+ /// fails and is retried.
+ ///
+ /// The data versions cannot tell a segment ZK metadata deleted and
recreated with the same name since it was read,
+ /// e.g. by an upload under the same name, from the one that was read. As
the IdealState is outside the property
+ /// store, the transaction uses the multi-write ZkClient directly instead of
[ZkMultiWriteBuilder].
+ public Set<String> removeInProgressSegmentsZKMetadataNotInIdealState(String
tableNameWithType,
+ Collection<String> segmentNames) {
+ return
removeInProgressSegmentsZKMetadataNotInIdealState(tableNameWithType,
segmentNames, () -> {
+ });
+ }
+
+ /// @param afterRead Runs after reading the IdealState and the segment ZK
metadata, before removing it
+ @VisibleForTesting
+ Set<String> removeInProgressSegmentsZKMetadataNotInIdealState(String
tableNameWithType,
+ Collection<String> segmentNames, Runnable afterRead) {
+ ZkClient zkClient = getOrBuildMultiWriteZkClient();
+ String idealStatePath = PropertyPathBuilder.idealState(_helixClusterName,
tableNameWithType);
+ String propertyStoreRoot =
PropertyPathBuilder.propertyStore(_helixClusterName);
+ for (int attempt = 1; ; attempt++) {
+ Stat idealStateStat = new Stat();
+ ZNRecord idealState = zkClient.readData(idealStatePath, idealStateStat);
Review Comment:
Done in df8842da90: a missing IdealState now returns an empty set without
removing anything. It handles both a null result and the ZkNoNodeException that
the ZkClient can throw for a missing node. The real-ZooKeeper test now also
covers a deleted table.
--
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]
