Jackie-Jiang commented on code in PR #20040:
URL: https://github.com/apache/pinot/pull/20040#discussion_r4393600976
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -2196,6 +2294,36 @@ && new
LLCSegmentName(segmentEntry.getKey()).getPartitionGroupId() == partitionI
}
}
+ if (!endOfLifeCandidates.isEmpty()) {
+ Set<Integer> endOfLifePartitionGroupIds =
fetchEndOfLifePartitionGroupIds(streamConfigs, endOfLifeCandidates);
+ for (SegmentZKMetadata segmentZKMetadata : endOfLifeCandidates) {
+ String segmentName = segmentZKMetadata.getSegmentName();
+ if (endOfLifePartitionGroupIds.contains(new
LLCSegmentName(segmentName).getPartitionGroupId())
+ && markEndOfPartitionGroup(realtimeTableName, segmentName)) {
+ segmentZKMetadata.setEndOfPartitionGroup(true);
+ if (onlineEndOfLifeCandidates.contains(segmentName)) {
+ sendPartitionEndOfLifeMessage(realtimeTableName, segmentName);
Review Comment:
[P2] Reconcile cleanup for already-marked partitions
The terminal mark is persisted before this best-effort notification, but
subsequent validation runs exclude marked segments at line 2210.
`MessagingServiceUtils.send()` catches exceptions and returns zero, and
`sendPartitionEndOfLifeMessage()` only logs the result. If delivery fails once,
or the controller stops between persistence and sending, servers already ONLINE
retain `LLC_PARTITION_CONSUMING=0` with no later cleanup attempt, preserving
the false alerts this change addresses. Reconcile notifications for marked
ONLINE segments or retain pending cleanup until it succeeds. Add a test where
the first send fails and a later validation run repairs the server metrics.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -949,10 +949,95 @@ private String createNewSegmentMetadata(TableConfig
tableConfig,
LOGGER.info(
"Skipping creation of new segment metadata after segment: {} during
commit. Reason: Partition ID: {} not "
+ "found in upstream metadata.", committingSegmentName,
committingSegmentPartitionGroupId);
+ if (fetchEndOfLifePartitionGroupIds(streamConfigs,
List.of(committingSegmentZKMetadata)).contains(
+ committingSegmentPartitionGroupId)) {
+ markEndOfPartitionGroup(realtimeTableName, committingSegmentName);
+ }
}
return newConsumingSegmentName;
}
+ /// Asks the stream which of the given completed segments are the last
segment of a partition group that reached end
+ /// of life (see [StreamMetadataProvider#fetchEndOfLifePartitionGroupIds]).
Returns their partition group ids.
+ @VisibleForTesting
+ Set<Integer> fetchEndOfLifePartitionGroupIds(List<StreamConfig>
streamConfigs,
+ Collection<SegmentZKMetadata> completedSegmentsZKMetadata) {
+ int numStreams = streamConfigs.size();
+ Map<Integer, List<PartitionGroupConsumptionStatus>>
streamConfigIndexToStatuses = new HashMap<>();
+ for (SegmentZKMetadata segmentZKMetadata : completedSegmentsZKMetadata) {
+ if (segmentZKMetadata.getEndOffset() == null) {
+ continue;
+ }
+ LLCSegmentName llcSegmentName = new
LLCSegmentName(segmentZKMetadata.getSegmentName());
+ int partitionGroupId = llcSegmentName.getPartitionGroupId();
+ // Same partition id translation as
getPartitionGroupConsumptionStatusList()
+ int streamConfigIndex =
+ numStreams == 1 ? 0 :
IngestionConfigUtils.getStreamConfigIndexFromPinotPartitionId(partitionGroupId);
+ if (streamConfigIndex >= numStreams) {
+ continue;
+ }
+ int streamPartitionId = numStreams == 1 ? partitionGroupId
+ :
IngestionConfigUtils.getStreamPartitionIdFromPinotPartitionId(partitionGroupId);
+ StreamPartitionMsgOffsetFactory offsetFactory =
+
StreamConsumerFactoryProvider.create(streamConfigs.get(streamConfigIndex)).createStreamMsgOffsetFactory();
+ streamConfigIndexToStatuses.computeIfAbsent(streamConfigIndex, k -> new
ArrayList<>()).add(
+ new PartitionGroupConsumptionStatus(partitionGroupId,
streamPartitionId, llcSegmentName.getSequenceNumber(),
+ offsetFactory.create(segmentZKMetadata.getStartOffset()),
+ offsetFactory.create(segmentZKMetadata.getEndOffset()),
segmentZKMetadata.getStatus().toString()));
+ }
+ Set<Integer> endOfLifePartitionGroupIds = new HashSet<>();
+ for (Map.Entry<Integer, List<PartitionGroupConsumptionStatus>> entry :
streamConfigIndexToStatuses.entrySet()) {
+ StreamConfig streamConfig = streamConfigs.get(entry.getKey());
+ try (StreamMetadataProvider metadataProvider =
StreamConsumerFactoryProvider.create(streamConfig)
+
.createStreamMetadataProvider(getTableTopicUniqueClientId(streamConfig))) {
+ endOfLifePartitionGroupIds.addAll(
+ metadataProvider.fetchEndOfLifePartitionGroupIds(streamConfig,
entry.getValue(),
+ (int) STREAM_FETCH_TIMEOUT_MS));
+ } catch (Exception e) {
+ LOGGER.warn("Failed to fetch end of life partition groups of topic: {}
for table: {}",
+ streamConfig.getTopicName(), streamConfig.getTableNameWithType(),
e);
+ }
+ }
+ return endOfLifePartitionGroupIds;
+ }
+
+ /// Marks the last segment of a partition group that reached end of life in
its ZK metadata, so that servers stop
+ /// reporting the partition as consuming, and sets the PARTITION_END_OF_LIFE
gauge. Returns whether it is marked.
+ private boolean markEndOfPartitionGroup(String realtimeTableName, String
segmentName) {
+ try {
+ Stat stat = new Stat();
+ SegmentZKMetadata segmentZKMetadata =
getSegmentZKMetadata(realtimeTableName, segmentName, stat);
+ if (!segmentZKMetadata.isEndOfPartitionGroup()) {
+ segmentZKMetadata.setEndOfPartitionGroup(true);
+ persistSegmentZKMetadata(realtimeTableName, segmentZKMetadata,
stat.getVersion());
Review Comment:
[P2] Preserve the timestamp used by pauseless commit recovery
This can rewrite a segment while it remains COMMITTING, advancing the ZK
modification time that `findNewSegmentsOfEarlierAttempts()` uses as the
commit-start bound. For example, an attempt creates successor metadata at 12:00
and fails its IdealState update; a retry at 12:02 observes the now-ended shard,
persists this flag, and fails before cleanup. The next retry searches from
12:01 and misses the original successor. Validation subsequently selects that
orphan as the latest segment and can add it as CONSUMING despite the partition
having ended, also emitting a false stream-data-loss count. Preserve a stable
commit-start timestamp or make successor discovery independent of this mutable
modification time. A regression test should cover marking terminal state
between failed pauseless attempts across minute boundaries.
--
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]
