xiangfu0 commented on code in PR #19740:
URL: https://github.com/apache/pinot/pull/19740#discussion_r4278274406
##########
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) {
Review Comment:
`isCommitStartResumable` already read the committing segment's ZK metadata a
moment earlier, and here you re-read it just for the Stat/mtime. Minor, but you
could thread the `Stat` through from the first read to save one ZK round-trip
per resume.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1398,6 +1564,11 @@ private Map<Integer, SegmentZKMetadata>
getLatestSegmentZKMetadataMap(String rea
} else {
if (llcSegmentName.getSequenceNumber() >
latestLLCSegmentName.getSequenceNumber()) {
return llcSegmentName;
+ } else if (llcSegmentName.getSequenceNumber() ==
latestLLCSegmentName.getSequenceNumber()) {
+ LOGGER.warn("Found segments: {} and {} with the same sequence
number in partition: {} of table: {}",
+ latestLLCSegmentName.getSegmentName(), segmentName,
partitionId, realtimeTableName);
+ return
!instanceStatesMap.containsKey(latestLLCSegmentName.getSegmentName())
Review Comment:
Nice tie-break. One edge case: when *neither* duplicate-seq segment is in
the IdealState this keeps whichever `getLLCSegments()` returned first, so the
pick is order-dependent. It's benign (both share the same seq, so the next
seq+1 name is identical either way), but a short comment that this branch is
intentionally best-effort for the "neither in IS" case would save the next
reader some head-scratching.
##########
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;
Review Comment:
The probe window's lower bound comes from the committing segment's ZK mtime.
That's safe today because while the status is COMMITTING nothing else rewrites
this znode (commit-end only runs once it's DONE, at which point resume is no
longer attempted). Might be worth a one-line comment noting that invariant,
since a future write to the committing metadata here would silently shrink the
window and let a duplicate new segment slip through.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1517,6 +1695,20 @@ IdealState updateIdealStateOnSegmentCompletion(String
realtimeTableName, String
false);
}
+ /// Returns `true` if the IdealState already reflects the completion of the
committing segment, i.e. all its replicas
+ /// are ONLINE, and the new consuming segment (if one is expected) is
present.
+ @VisibleForTesting
+ static boolean isSegmentCompletionApplied(IdealState idealState, String
committingSegmentName,
+ @Nullable String newSegmentName) {
+ Map<String, String> committingSegmentInstanceStateMap =
idealState.getInstanceStateMap(committingSegmentName);
+ if (committingSegmentInstanceStateMap == null ||
committingSegmentInstanceStateMap.isEmpty()
+ ||
!committingSegmentInstanceStateMap.values().stream().allMatch(SegmentStateModel.ONLINE::equals))
{
Review Comment:
`allMatch(ONLINE)` treats the completion as applied only when every replica
is ONLINE. If a replica is transiently OFFLINE/ERROR this returns false and we
re-run the update, which then hits the `containsValue(CONSUMING)` precondition.
That matches prior behavior so it's fine, just flagging that the "applied"
check is strictly all-ONLINE and won't short-circuit a partially-ONLINE
committing segment.
##########
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) {
Review Comment:
`isCommitStartResumable` already read the committing segment's ZK metadata a
moment earlier, and here you re-read it just for the Stat/mtime. Minor, but you
could thread the `Stat` through from the first read to save one ZK round-trip
per resume.
##########
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;
Review Comment:
The probe window's lower bound comes from the committing segment's ZK mtime.
That's safe today because while the status is COMMITTING nothing else rewrites
this znode (commit-end only runs once it's DONE, at which point resume is no
longer attempted). Might be worth a one-line comment noting that invariant,
since a future write to the committing metadata here would silently shrink the
window and let a duplicate new segment slip through.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1517,6 +1695,20 @@ IdealState updateIdealStateOnSegmentCompletion(String
realtimeTableName, String
false);
}
+ /// Returns `true` if the IdealState already reflects the completion of the
committing segment, i.e. all its replicas
+ /// are ONLINE, and the new consuming segment (if one is expected) is
present.
+ @VisibleForTesting
+ static boolean isSegmentCompletionApplied(IdealState idealState, String
committingSegmentName,
+ @Nullable String newSegmentName) {
+ Map<String, String> committingSegmentInstanceStateMap =
idealState.getInstanceStateMap(committingSegmentName);
+ if (committingSegmentInstanceStateMap == null ||
committingSegmentInstanceStateMap.isEmpty()
+ ||
!committingSegmentInstanceStateMap.values().stream().allMatch(SegmentStateModel.ONLINE::equals))
{
Review Comment:
`allMatch(ONLINE)` treats the completion as applied only when every replica
is ONLINE. If a replica is transiently OFFLINE/ERROR this returns false and we
re-run the update, which then hits the `containsValue(CONSUMING)` precondition.
That matches prior behavior so it's fine, just flagging that the "applied"
check is strictly all-ONLINE and won't short-circuit a partially-ONLINE
committing segment.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1398,6 +1564,11 @@ private Map<Integer, SegmentZKMetadata>
getLatestSegmentZKMetadataMap(String rea
} else {
if (llcSegmentName.getSequenceNumber() >
latestLLCSegmentName.getSequenceNumber()) {
return llcSegmentName;
+ } else if (llcSegmentName.getSequenceNumber() ==
latestLLCSegmentName.getSequenceNumber()) {
+ LOGGER.warn("Found segments: {} and {} with the same sequence
number in partition: {} of table: {}",
+ latestLLCSegmentName.getSegmentName(), segmentName,
partitionId, realtimeTableName);
+ return
!instanceStatesMap.containsKey(latestLLCSegmentName.getSegmentName())
Review Comment:
Nice tie-break. One edge case: when *neither* duplicate-seq segment is in
the IdealState this keeps whichever `getLLCSegments()` returned first, so the
pick is order-dependent. It's benign (both share the same seq, so the next
seq+1 name is identical either way), but a short comment that this branch is
intentionally best-effort for the "neither in IS" case would save the next
reader some head-scratching.
--
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]
