suvodeep-pyne commented on code in PR #19740:
URL: https://github.com/apache/pinot/pull/19740#discussion_r4341255425
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -836,17 +904,35 @@ private String createNewSegmentMetadata(TableConfig
tableConfig,
+ "paused.", committingSegmentName);
return null;
}
+ // Another attempt of the same commit start might have created the new
segment already, and the segment might be
+ // consuming or even committed, so never overwrite its ZK metadata
+ if (resumed) {
+ String existingNewSegmentName =
findNewSegmentOfEarlierAttempt(realtimeTableName, committingLLCSegment);
+ if (existingNewSegmentName != null) {
+ LOGGER.info("Using existing segment metadata for new segment: {}",
existingNewSegmentName);
+ return existingNewSegmentName;
Review Comment:
Good catch, thanks. Fixed in #19995. A resumed commit start now looks up the
earlier attempt's new segment before the pause and partition checks, uses it
only when those checks pass, and removes its ZK metadata once the commit
completes without it. Added failure → pause → retry tests, including validation
after resuming the table with an offset criteria. I also reproduced it end to
end on a local cluster: without the fix, resuming with consumeFrom=largest
added the kept segments from the committed end offset.
One note on the paused topic part: pauseless tables only support a single
stream, and validation does not skip a paused topic of a single-stream table,
so for topic pause the fix only removes the kept segment.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -806,11 +820,65 @@ 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. This relies on nothing
rewriting the committing segment ZK metadata
+ /// while it stays COMMITTING, as a later modification time would skip the
minutes in which the new segment was
+ /// created.
+ @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) {
+ String segmentName = new
LLCSegmentName(committingLLCSegment.getTableName(),
+ committingLLCSegment.getPartitionGroupId(),
committingLLCSegment.getSequenceNumber() + 1,
+ timeMs).getSegmentName();
+ if (segmentZKMetadataExists(realtimeTableName, segmentName)) {
+ return segmentName;
+ }
+ }
+ return null;
+ }
+
// Step 2: Create new segment metadata
@Nullable
private String createNewSegmentMetadata(TableConfig tableConfig,
CommittingSegmentDescriptor committingSegmentDescriptor,
SegmentZKMetadata committingSegmentZKMetadata,
- InstancePartitions instancePartitions) {
+ InstancePartitions instancePartitions, boolean createIfAbsent, boolean
resumed) {
Review Comment:
Done in #19995: the lookup moved out of createNewSegmentMetadata into the
commit flow, and createNewSegmentMetadata takes the found segment instead of
the resumed flag, after the partition and pause checks.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -659,51 +659,65 @@ private void commitSegmentMetadataInternal(String
realtimeTableName,
// Step-1: Update PROPERTYSTORE
LOGGER.info("Committing segment metadata for segment: {}",
committingSegmentName);
long startTimeNs1 = System.nanoTime();
- SegmentZKMetadata committingSegmentZKMetadata = toCommitting
- ? updateSegmentZKMetadataToCommitting(realtimeTableName,
committingSegmentDescriptor)
- : updateSegmentZKMetadataToDone(realtimeTableName,
committingSegmentDescriptor, Status.IN_PROGRESS);
+ boolean resumed = toCommitting &&
isCommitStartResumable(realtimeTableName, committingSegmentDescriptor);
+ SegmentZKMetadata committingSegmentZKMetadata;
+ if (resumed) {
+ committingSegmentZKMetadata = getSegmentZKMetadata(realtimeTableName,
committingSegmentName);
Review Comment:
I kept this out of #19995 to keep the fix focused. It would change step 1 of
every pauseless commit to save a few reads on the resume path, which only runs
after a failed commit start. Happy to do it as a separate cleanup if you prefer.
--
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]
