Jackie-Jiang commented on code in PR #19740:
URL: https://github.com/apache/pinot/pull/19740#discussion_r4331476685
##########
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:
[P1] Find the retained successor before handling a pause
If the initial commit start creates successor N and its IdealState update
fails, the new failure handler retains N. If the table/topic is paused before
the same-offset retry, the pause check above returns null before this lookup.
The retry can then successfully mark the predecessor ONLINE, but the
post-update cleanup never sees N because newConsumingSegmentName is null.
The validator later selects N as the latest segment and takes the
metadata-without-IdealState repair branch. That branch neither checks whether
its topic is paused nor applies the requested resume offsetCriteria: it can
restart a paused topic after the completion timeout, or ignore consumeFrom when
a paused table is resumed.
Please resolve the retained successor before the pause decision so
successful paused completion can clean up its unused metadata, while preserving
the upstream partition-eligibility check. A failure → pause → retry →
validation regression test would cover this.
##########
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:
Optional simplification: combine resumability checking with the COMMITTING
transition
A prepareCommitStart helper could read SegmentZKMetadata and Stat once,
accept a valid same-offset COMMITTING state or perform the existing
version-checked IN_PROGRESS → COMMITTING transition, and return the metadata
plus resume information. The saved modification time could also supply the
successor probe's lower bound.
This would remove repeated metadata reads and the current flow where
isCommitStartResumable returns false for an invalid COMMITTING state only for
updateSegmentZKMetadataToCommitting to read it again and reject it. Preserve
the original modification time when resuming and keep the fresh deadline check
inside the IdealState updater.
##########
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:
Optional simplification: separate successor recovery from fresh creation
Keep recovery of the earlier attempt's successor in the pauseless
preparation flow, and let createNewSegmentMetadata handle fresh creation when
necessary. This would remove its resumed parameter and make the retained
segment's identity available for cleanup even when the retry observes a pause,
addressing the ordering issue in the other comment.
Preserve partition eligibility: finding an existing successor must not by
itself authorize assigning it after its upstream partition disappears. I would
retain atomic create-if-absent, the idempotent IdealState updater, and the
bounded timestamp probe. Replacing the probe with a persisted successor
identity would require a larger metadata-protocol change.
--
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]
