suvodeep-pyne commented on code in PR #19739:
URL: https://github.com/apache/pinot/pull/19739#discussion_r4328959340
##########
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:
Good catch, thanks. Fixed in 14879355b7. `getLatestSegmentZKMetadataMap` now
prefers the segment in the IdealState when several segments of a partition have
the highest sequence number. In your scenario the retry picks the successor A
published and processes it normally: it is CONSUMING, so nothing changes for
the partition, and the predecessor is never touched.
`testEnsureAllPartitionsConsumingRetryKeepsSegmentAddedByAnotherController`
covers both listing orders. Without the tie-break, the reversed order fails
with `PermanentUpdaterException: Segment testTable__0__1__...T1848Z is a
duplicate of existing segment testTable__0__1__...T1847Z`.
When neither successor is in the IdealState yet, whichever is listed first
is used. If that is the other controller's segment, the retry waits for the
completion timeout as before. If it is its own, the retry adds it, and the
other controller's next attempt then finds that segment in the IdealState.
#19740 adds the same tie-break for duplicates left by a failed commit start, so
whichever PR merges second will be rebased onto it.
##########
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:
Agreed that it is not accessed concurrently. I kept `ConcurrentHashMap`
anyway, so the map stays safe without relying on how `IdealStateGroupCommit`
hands the callback between threads. It holds at most one entry per partition
repaired in the run, so the cost is negligible.
##########
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:
Added
`testEnsureAllPartitionsConsumingAddsSegmentsCreatedByAttemptWithFailedIdealStateWrite`.
It calls the public `ensureAllPartitionsConsuming` through the real
`IdealStateGroupCommit` retry loop with a mocked `HelixDataAccessor`. The first
write fails because another controller uploads a segment in the meantime, and
the retry succeeds. The test checks that the new segment is added right away,
that no extra metadata is created, and that the uploaded segment is kept.
##########
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:
Added a new partition (partition 4) to
`testEnsureAllPartitionsConsumingRetryAddsSegmentsCreatedByEarlierAttempt`. The
retry adds the segment created by the first attempt as CONSUMING without
creating any other metadata for the partition, and `LLC_STREAM_DATA_LOSS` is
not reported.
--
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]
