xiangfu0 commented on code in PR #19740:
URL: https://github.com/apache/pinot/pull/19740#discussion_r4290462266
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -806,11 +820,63 @@ private SegmentZKMetadata
updateSegmentZKMetadataToCommitting(String realtimeTab
return segmentZKMetadata;
}
+ /// Returns `true` if a pauseless commit start can resume an earlier attempt
that moved the segment to COMMITTING
+ /// with the same end offset, but did not complete, e.g. because the
IdealState update failed or the controller
+ /// stopped. Steps 2 and 3 are idempotent for the same segment and end
offset: the resumed attempt uses the new
+ /// segment created by an earlier attempt if any (see
findNewSegmentOfEarlierAttempt), new segment ZK metadata is only
+ /// created if absent, and the IdealState update is skipped when it is
already applied. Once the max segment
+ /// completion time is exceeded, RealtimeSegmentValidationManager repairs
the segment instead.
+ private boolean isCommitStartResumable(String realtimeTableName,
+ CommittingSegmentDescriptor committingSegmentDescriptor) {
+ String segmentName = committingSegmentDescriptor.getSegmentName();
+ SegmentZKMetadata segmentZKMetadata =
getSegmentZKMetadata(realtimeTableName, segmentName);
+ if (segmentZKMetadata.getStatus() != Status.COMMITTING) {
+ return false;
+ }
+ String endOffset = segmentZKMetadata.getEndOffset();
+ if (!committingSegmentDescriptor.getNextOffset().equals(endOffset)) {
+ LOGGER.warn("Cannot resume commit start for segment: {} with end offset:
{}, segment is COMMITTING with end "
+ + "offset: {}", segmentName,
committingSegmentDescriptor.getNextOffset(), endOffset);
+ return false;
+ }
+ if (isExceededMaxSegmentCompletionTime(realtimeTableName, segmentName,
getCurrentTimeMs())) {
+ LOGGER.warn("Cannot resume commit start for segment: {} after the max
segment completion time", segmentName);
+ return false;
+ }
+ LOGGER.info("Resuming commit start for segment: {} which is COMMITTING
with end offset: {}", segmentName,
+ endOffset);
+ return true;
+ }
+
+ /// Returns the new segment created by an earlier attempt of a resumed
commit start, or `null` if there is none.
+ ///
+ /// The new segment is named after its creation time with minute
granularity, and is created between step 1 of the
+ /// first attempt (the modification time of the committing segment ZK
metadata) and now, which is bounded by the max
+ /// segment completion time. Probes the names for every minute in between,
with one minute of slack on both ends for
+ /// clock differences between controllers and ZK.
+ @Nullable
+ private String findNewSegmentOfEarlierAttempt(String realtimeTableName,
LLCSegmentName committingLLCSegment) {
+ Stat stat = new Stat();
+ getSegmentZKMetadata(realtimeTableName,
committingLLCSegment.getSegmentName(), stat);
+ long oneMinuteMs = TimeUnit.MINUTES.toMillis(1);
+ long startTimeMs = Math.floorDiv(stat.getMtime(), oneMinuteMs) *
oneMinuteMs - oneMinuteMs;
+ long endTimeMs = Math.floorDiv(getCurrentTimeMs(), oneMinuteMs) *
oneMinuteMs + oneMinuteMs;
Review Comment:
The probe window derives `startTimeMs` from the committing segment's ZK
mtime but `endTimeMs` from this controller's clock, while the earlier attempt
named its new segment using its own clock. If the earlier controller led both
ZK and this one by more than the 1-minute slack, the probe would miss that
segment and a second new segment gets created. It degrades gracefully (the
losing attempt fails the CONSUMING precondition and cleans up its orphan), so
this is minor — but a short comment noting the clock-skew assumption would help
future readers.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -806,11 +820,63 @@ private SegmentZKMetadata
updateSegmentZKMetadataToCommitting(String realtimeTab
return segmentZKMetadata;
}
+ /// Returns `true` if a pauseless commit start can resume an earlier attempt
that moved the segment to COMMITTING
+ /// with the same end offset, but did not complete, e.g. because the
IdealState update failed or the controller
+ /// stopped. Steps 2 and 3 are idempotent for the same segment and end
offset: the resumed attempt uses the new
+ /// segment created by an earlier attempt if any (see
findNewSegmentOfEarlierAttempt), new segment ZK metadata is only
+ /// created if absent, and the IdealState update is skipped when it is
already applied. Once the max segment
+ /// completion time is exceeded, RealtimeSegmentValidationManager repairs
the segment instead.
+ private boolean isCommitStartResumable(String realtimeTableName,
+ CommittingSegmentDescriptor committingSegmentDescriptor) {
+ String segmentName = committingSegmentDescriptor.getSegmentName();
+ SegmentZKMetadata segmentZKMetadata =
getSegmentZKMetadata(realtimeTableName, segmentName);
+ if (segmentZKMetadata.getStatus() != Status.COMMITTING) {
+ return false;
+ }
+ String endOffset = segmentZKMetadata.getEndOffset();
+ if (!committingSegmentDescriptor.getNextOffset().equals(endOffset)) {
+ LOGGER.warn("Cannot resume commit start for segment: {} with end offset:
{}, segment is COMMITTING with end "
+ + "offset: {}", segmentName,
committingSegmentDescriptor.getNextOffset(), endOffset);
+ return false;
+ }
+ if (isExceededMaxSegmentCompletionTime(realtimeTableName, segmentName,
getCurrentTimeMs())) {
+ LOGGER.warn("Cannot resume commit start for segment: {} after the max
segment completion time", segmentName);
+ return false;
+ }
+ LOGGER.info("Resuming commit start for segment: {} which is COMMITTING
with end offset: {}", segmentName,
+ endOffset);
+ return true;
+ }
+
+ /// Returns the new segment created by an earlier attempt of a resumed
commit start, or `null` if there is none.
+ ///
+ /// The new segment is named after its creation time with minute
granularity, and is created between step 1 of the
+ /// first attempt (the modification time of the committing segment ZK
metadata) and now, which is bounded by the max
+ /// segment completion time. Probes the names for every minute in between,
with one minute of slack on both ends for
+ /// clock differences between controllers and ZK.
+ @Nullable
+ private String findNewSegmentOfEarlierAttempt(String realtimeTableName,
LLCSegmentName committingLLCSegment) {
+ Stat stat = new Stat();
+ getSegmentZKMetadata(realtimeTableName,
committingLLCSegment.getSegmentName(), stat);
+ long oneMinuteMs = TimeUnit.MINUTES.toMillis(1);
+ long startTimeMs = Math.floorDiv(stat.getMtime(), oneMinuteMs) *
oneMinuteMs - oneMinuteMs;
+ long endTimeMs = Math.floorDiv(getCurrentTimeMs(), oneMinuteMs) *
oneMinuteMs + oneMinuteMs;
+ for (long timeMs = startTimeMs; timeMs <= endTimeMs; timeMs +=
oneMinuteMs) {
Review Comment:
This loop does one ZK `exists` round-trip per minute between the committing
segment's mtime and now. With the default completion time that's only a handful
of calls on a rare failure path, but `max.segment.completion.time.millis` is
operator-configurable and unbounded, so a large value turns the resume into
many serial ZK reads. Might be worth capping the window explicitly at the
max-completion duration.
--
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]
