Jackie-Jiang commented on code in PR #19739:
URL: https://github.com/apache/pinot/pull/19739#discussion_r4328748726
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1938,6 +1962,7 @@ && new
LLCSegmentName(segmentEntry.getKey()).getPartitionGroupId() == partitionI
String newSegmentName =
setupNewPartitionGroup(tableConfig,
streamMetadata.getStreamConfig(), partitionGroupMetadata,
currentTimeMs, instancePartitions, numPartitions,
numReplicas);
+ newSegmentsCreated.put(newSegmentName, false);
Review Comment:
Optional test coverage: the new retry tests cover a stuck commit and an
all-OFFLINE partition, but neither exercises this separate new-partition
creation hook. Add a case where the first attempt creates a new partition's
segment metadata and its IdealState write is discarded, then retry with the
same context before the completion timeout. Assert that the same segment and
start offset are reused, exactly one successor is added as CONSUMING, no
additional metadata is created, and LLC_STREAM_DATA_LOSS is not reported.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1448,6 +1448,10 @@ public void ensureAllPartitionsConsuming(TableConfig
tableConfig, List<StreamCon
Preconditions.checkState(!_isStopping, "Segment manager is stopping");
String realtimeTableName = tableConfig.getTableName();
+ // The updater below runs again when the IdealState write fails (e.g. on a
version conflict), while the segment ZK
+ // metadata created by the failed attempt remains. Track the created
segments so that the next attempt still adds
+ // them to the IdealState.
+ Map<String, Boolean> newSegmentsCreated = new ConcurrentHashMap<>();
Review Comment:
Optional simplification: a HashMap is sufficient here. IdealStateGroupCommit
safely publishes the callback through its concurrent queue, invokes callbacks
under a single queue leader, and holds that ownership across the synchronous
RetryPolicy.attempt() loop. Although the callback can run on another request
thread, this invocation's map is not accessed concurrently. ConcurrentHashMap
adds no cross-controller protection. Keeping the Boolean value makes sense
because it preserves the missing-CONSUMING-predecessor metric distinction.
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java:
##########
@@ -1025,6 +1026,82 @@ public void testPartialOfflineReplicaRepair() {
assertEquals(consumingSegmentInstanceStateMap.get(offlineInstance),
SegmentStateModel.CONSUMING);
}
+ @Test
+ public void
testEnsureAllPartitionsConsumingRetryAddsSegmentsCreatedByEarlierAttempt() {
+ ControllerMetrics controllerMetrics = mock(ControllerMetrics.class);
+ FakePinotLLCRealtimeSegmentManager segmentManager = new
FakePinotLLCRealtimeSegmentManager(controllerMetrics);
+ setUpNewTable(segmentManager, 2, 5, 4);
+ Map<String, Map<String, String>> instanceStatesMap =
segmentManager._idealState.getRecord().getMapFields();
+
+ // Partition 0: step 1 of the segment commit is done, but the segment is
still CONSUMING in IdealState
+ String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0,
CURRENT_TIME_MS).getSegmentName();
+ SegmentZKMetadata committingSegmentZKMetadata =
segmentManager._segmentZKMetadataMap.get(committingSegment);
+ committingSegmentZKMetadata.setStatus(Status.DONE);
+ committingSegmentZKMetadata.setEndOffset(NEXT_OFFSET);
+
segmentManager._segmentsExceedingMaxSegmentCompletionTime.add(committingSegment);
+ // Partition 1: all replicas of the consuming segment are OFFLINE
+ String offlineSegment = new LLCSegmentName(RAW_TABLE_NAME, 1, 0,
CURRENT_TIME_MS).getSegmentName();
+ instanceStatesMap.get(offlineSegment).replaceAll((instance, state) ->
SegmentStateModel.OFFLINE);
+
+ // The first attempt creates the new segments, but its IdealState write
fails
+ Map<String, Boolean> newSegmentsCreated = new HashMap<>();
+ runFailedEnsureAllPartitionsConsumingAttempt(segmentManager,
newSegmentsCreated);
+ String newSegment0 = new LLCSegmentName(RAW_TABLE_NAME, 0, 1,
CURRENT_TIME_MS).getSegmentName();
+ String newSegment1 = new LLCSegmentName(RAW_TABLE_NAME, 1, 1,
CURRENT_TIME_MS).getSegmentName();
+
+ // The retry starts from the unchanged IdealState. It must add the new
segments right away instead of waiting for
+ // them to exceed the max segment completion time.
+ segmentManager.ensureAllPartitionsConsuming(segmentManager._tableConfig,
segmentManager._streamConfigs,
+ segmentManager._idealState, getStreamMetadataList(segmentManager),
null, newSegmentsCreated);
+ assertEquals(segmentManager._segmentZKMetadataMap.keySet().stream()
+ .filter(segment -> new LLCSegmentName(segment).getPartitionGroupId()
<= 1)
+ .collect(Collectors.toSet()), Set.of(committingSegment, newSegment0,
offlineSegment, newSegment1));
+ assertEquals(new
HashSet<>(instanceStatesMap.get(committingSegment).values()),
Set.of(SegmentStateModel.ONLINE));
+ assertEquals(new HashSet<>(instanceStatesMap.get(newSegment0).values()),
Set.of(SegmentStateModel.CONSUMING));
+ assertEquals(new
HashSet<>(instanceStatesMap.get(offlineSegment).values()),
Set.of(SegmentStateModel.OFFLINE));
+ assertEquals(new HashSet<>(instanceStatesMap.get(newSegment1).values()),
Set.of(SegmentStateModel.CONSUMING));
+ verify(controllerMetrics,
never()).addMeteredTableValue(REALTIME_TABLE_NAME,
ControllerMeter.LLC_STREAM_DATA_LOSS,
+ 1L);
+ }
+
+ @Test
+ public void
testEnsureAllPartitionsConsumingRetryReportsDataLossWhenReplacedSegmentGoesOffline()
{
+ ControllerMetrics controllerMetrics = mock(ControllerMetrics.class);
+ FakePinotLLCRealtimeSegmentManager segmentManager = new
FakePinotLLCRealtimeSegmentManager(controllerMetrics);
+ setUpNewTable(segmentManager, 2, 5, 4);
+ Map<String, Map<String, String>> instanceStatesMap =
segmentManager._idealState.getRecord().getMapFields();
+ String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0,
CURRENT_TIME_MS).getSegmentName();
+ SegmentZKMetadata committingSegmentZKMetadata =
segmentManager._segmentZKMetadataMap.get(committingSegment);
+ committingSegmentZKMetadata.setStatus(Status.DONE);
+ committingSegmentZKMetadata.setEndOffset(NEXT_OFFSET);
+
segmentManager._segmentsExceedingMaxSegmentCompletionTime.add(committingSegment);
+
+ Map<String, Boolean> newSegmentsCreated = new HashMap<>();
+ runFailedEnsureAllPartitionsConsumingAttempt(segmentManager,
newSegmentsCreated);
+
+ // All replicas of the segment being replaced go OFFLINE before the retry
+ instanceStatesMap.get(committingSegment).replaceAll((instance, state) ->
SegmentStateModel.OFFLINE);
+ segmentManager.ensureAllPartitionsConsuming(segmentManager._tableConfig,
segmentManager._streamConfigs,
+ segmentManager._idealState, getStreamMetadataList(segmentManager),
null, newSegmentsCreated);
+ String newSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 1,
CURRENT_TIME_MS).getSegmentName();
+ assertEquals(new HashSet<>(instanceStatesMap.get(newSegment).values()),
Set.of(SegmentStateModel.CONSUMING));
+ verify(controllerMetrics).addMeteredTableValue(REALTIME_TABLE_NAME,
ControllerMeter.LLC_STREAM_DATA_LOSS, 1L);
+ }
+
+ /// Runs ensureAllPartitionsConsuming on a copy of the IdealState, as an
attempt whose IdealState write fails.
+ private void
runFailedEnsureAllPartitionsConsumingAttempt(FakePinotLLCRealtimeSegmentManager
segmentManager,
+ Map<String, Boolean> newSegmentsCreated) {
+ IdealState failedAttemptIdealState = new IdealState(REALTIME_TABLE_NAME);
+ failedAttemptIdealState.getRecord()
+
.setMapFields(cloneInstanceStatesMap(segmentManager._idealState.getRecord().getMapFields()));
+ segmentManager.ensureAllPartitionsConsuming(segmentManager._tableConfig,
segmentManager._streamConfigs,
+ failedAttemptIdealState, getStreamMetadataList(segmentManager), null,
newSegmentsCreated);
Review Comment:
Optional test coverage: add a test through the public
ensureAllPartitionsConsuming() entry point where the first IdealState write
fails and the next succeeds. This helper manually passes the same map between
attempts, so the current tests verify the recovery logic but would still pass
if production accidentally allocated the map inside the callback or stopped
forwarding it. A public-path retry test should verify that the previously
created segments are added immediately without creating additional metadata and
that concurrent unrelated IdealState changes are preserved.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1899,11 +1914,19 @@ IdealState ensureAllPartitionsConsuming(TableConfig
tableConfig, List<StreamConf
// i.e. after updating old segment metadata (old segment metadata
state = DONE/COMMITTING)
// and creating new segment metadata (new segment metadata state =
IN_PROGRESS),
// but before updating ideal state (new segment ideal missing from
ideal state)
- if (!isExceededMaxSegmentCompletionTime(realtimeTableName,
latestSegmentName, currentTimeMs)) {
- continue;
+ // The segment can also be created by an earlier attempt of this
update, which is not committing anything and
+ // can be added right away.
+ Boolean replacesConsumingSegment =
newSegmentsCreated.get(latestSegmentName);
+ boolean createdByEarlierAttempt = replacesConsumingSegment != null;
+ if (createdByEarlierAttempt) {
+ LOGGER.info("Adding segment: {} created by an earlier attempt of the
IdealState update", latestSegmentName);
+ } else {
+ if (!isExceededMaxSegmentCompletionTime(realtimeTableName,
latestSegmentName, currentTimeMs)) {
Review Comment:
[P2] Reconcile a competing published successor before bypassing the timeout
During a leadership handoff, the old controller can still be repairing this
table: ControllerPeriodicTask snapshots leadership before processing, and
losing leadership does not cancel that in-flight work. Both controllers can
read segment N and create timestamp-distinct successors with the same
partition/sequence. If A publishes first, B loses the IdealState CAS and
retries. getLatestSegmentZKMetadataMap() has no IdealState-aware tie break, so
B can select its own unpublished successor. This new exemption then reaches the
duplicate-sequence guard in updateInstanceStatesForNewConsumingSegment(), which
throws PermanentUpdaterException and aborts the repair and other processed
entries in the same group-commit batch. Before this change, B skipped the fresh
orphan, allowing remaining updates to proceed.
Prefer the segment already present in the current IdealState among equal
highest-sequence metadata candidates, and process that published segment
normally. Add a regression covering both metadata enumeration orders. Resolve
the conflict before mutating predecessor state; simply swallowing the duplicate
exception is unsafe because the helper changes the previous CONSUMING segment
to ONLINE before the duplicate check. The underlying duplicate-metadata race
predates this PR; the immediate skip-to-batch-failure behavior is the new edge
case.
--
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]
