suvodeep-pyne commented on code in PR #19909:
URL: https://github.com/apache/pinot/pull/19909#discussion_r4339390335
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -2956,22 +2958,14 @@ public void repairSegmentsInErrorState(TableConfig
tableConfig,
boolean repairCommittingSegments =
Review Comment:
Agreed. In a779ff97dd, `allowRepairOfCommittingSegments` is evaluated lazily
(memoized), only when a segment needs a gated repair, so its WARN only appears
next to the per-segment skip WARN.
The rationale is now in one place, the method's Javadoc. The case comments
only say what each case does, and the PR's private `resetErrorReplicas` helper
and its doc are gone.
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java:
##########
@@ -2315,6 +2318,118 @@ public void
testShouldRepairErrorSegmentsForPartialUpsertOrDedup() {
pinotLLCRealtimeSegmentManager.shouldRepairErrorSegmentsForPartialUpsertOrDedup(DisasterRecoveryMode.ALWAYS));
}
+ @Test
+ public void
testRepairSegmentsInErrorStateResetsOnlyErrorReplicasForPauselessDedupTable()
+ throws InvalidConfigException {
+ PinotHelixResourceManager mockHelixResourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics mockControllerMetrics = mock(ControllerMetrics.class);
+ FakePinotLLCRealtimeSegmentManager segmentManager =
+ new FakePinotLLCRealtimeSegmentManager(mockHelixResourceManager,
mockControllerMetrics);
+ TableConfig tableConfig = createPauselessDedupTableConfig();
+
+ String server0 = Helix.PREFIX_OF_SERVER_INSTANCE + 0;
+ String server1 = Helix.PREFIX_OF_SERVER_INSTANCE + 1;
+ String committingPartialErrorSegment = new LLCSegmentName(RAW_TABLE_NAME,
0, 0, CURRENT_TIME_MS).getSegmentName();
+ String committingAllErrorSegment = new LLCSegmentName(RAW_TABLE_NAME, 0,
1, CURRENT_TIME_MS).getSegmentName();
+ String doneSegment = new LLCSegmentName(RAW_TABLE_NAME, 1, 0,
CURRENT_TIME_MS).getSegmentName();
+ String inProgressSegment = new LLCSegmentName(RAW_TABLE_NAME, 1, 1,
CURRENT_TIME_MS).getSegmentName();
+
+ IdealState idealState = new IdealState(REALTIME_TABLE_NAME);
+ ExternalView externalView = new ExternalView(REALTIME_TABLE_NAME);
+ for (String segment : List.of(committingPartialErrorSegment,
committingAllErrorSegment, doneSegment,
+ inProgressSegment)) {
+ idealState.setPartitionState(segment, server0, SegmentStateModel.ONLINE);
+ idealState.setPartitionState(segment, server1, SegmentStateModel.ONLINE);
+ }
+ externalView.setState(committingPartialErrorSegment, server0,
SegmentStateModel.ONLINE);
+ externalView.setState(committingPartialErrorSegment, server1,
SegmentStateModel.ERROR);
+ externalView.setState(committingAllErrorSegment, server0,
SegmentStateModel.ERROR);
+ externalView.setState(committingAllErrorSegment, server1,
SegmentStateModel.ERROR);
+ externalView.setState(doneSegment, server0, SegmentStateModel.ONLINE);
+ externalView.setState(doneSegment, server1, SegmentStateModel.ERROR);
+ externalView.setState(inProgressSegment, server0, SegmentStateModel.ERROR);
+ externalView.setState(inProgressSegment, server1, SegmentStateModel.ERROR);
+ segmentManager._idealState = idealState;
+
when(mockHelixResourceManager.getTableExternalView(REALTIME_TABLE_NAME)).thenReturn(externalView);
+ mockSegmentZKMetadataStatus(mockHelixResourceManager,
committingPartialErrorSegment, Status.COMMITTING);
+ mockSegmentZKMetadataStatus(mockHelixResourceManager,
committingAllErrorSegment, Status.COMMITTING);
+ mockSegmentZKMetadataStatus(mockHelixResourceManager, doneSegment,
Status.DONE);
+ mockSegmentZKMetadataStatus(mockHelixResourceManager, inProgressSegment,
Status.IN_PROGRESS);
+
+ segmentManager.repairSegmentsInErrorState(tableConfig, false);
+
+ // ERROR replicas are reset even though dedup is enabled, since a reset
only reloads data a peer or the deep
+ // store already holds
+ verify(mockHelixResourceManager).resetSegment(REALTIME_TABLE_NAME,
committingPartialErrorSegment, server1);
+ verify(mockHelixResourceManager).resetSegment(REALTIME_TABLE_NAME,
doneSegment, server1);
+ verify(mockHelixResourceManager).resetSegment(REALTIME_TABLE_NAME,
inProgressSegment, server0);
+ verify(mockHelixResourceManager).resetSegment(REALTIME_TABLE_NAME,
inProgressSegment, server1);
+ // Healthy replicas are never reset, and no untargeted (all-replica) reset
is issued
+ verify(mockHelixResourceManager,
never()).resetSegment(eq(REALTIME_TABLE_NAME), anyString(), isNull());
+ verify(mockHelixResourceManager,
never()).resetSegment(REALTIME_TABLE_NAME, committingPartialErrorSegment,
+ server0);
+ verify(mockHelixResourceManager,
never()).resetSegment(REALTIME_TABLE_NAME, doneSegment, server0);
+ // The COMMITTING segment with all replicas in ERROR is neither reset nor
re-ingested without the override
+ verify(mockHelixResourceManager,
never()).resetSegment(eq(REALTIME_TABLE_NAME), eq(committingAllErrorSegment),
+ any());
+ verify(mockHelixResourceManager,
never()).getDataInstanceAdminEndpoints(any());
+ // Every segment with an ERROR replica is counted, and only the segment
left unrepaired is unrecoverable
+ verify(mockControllerMetrics).setOrUpdateTableGauge(REALTIME_TABLE_NAME,
+ ControllerGauge.PAUSELESS_SEGMENTS_IN_ERROR_COUNT, 4L);
+ verify(mockControllerMetrics).setOrUpdateTableGauge(REALTIME_TABLE_NAME,
+ ControllerGauge.PAUSELESS_SEGMENTS_IN_UNRECOVERABLE_ERROR_COUNT, 1L);
+ }
+
+ @Test
+ public void
testRepairSegmentsInErrorStateReingestsCommittingSegmentForDedupTableWithOverride()
+ throws InvalidConfigException {
+ PinotHelixResourceManager mockHelixResourceManager =
mock(PinotHelixResourceManager.class);
+ FakePinotLLCRealtimeSegmentManager segmentManager =
+ new FakePinotLLCRealtimeSegmentManager(mockHelixResourceManager);
+ TableConfig tableConfig = createPauselessDedupTableConfig();
+
+ String server0 = Helix.PREFIX_OF_SERVER_INSTANCE + 0;
+ String server1 = Helix.PREFIX_OF_SERVER_INSTANCE + 1;
+ String committingAllErrorSegment = new LLCSegmentName(RAW_TABLE_NAME, 0,
0, CURRENT_TIME_MS).getSegmentName();
+
+ IdealState idealState = new IdealState(REALTIME_TABLE_NAME);
+ idealState.setPartitionState(committingAllErrorSegment, server0,
SegmentStateModel.ONLINE);
+ idealState.setPartitionState(committingAllErrorSegment, server1,
SegmentStateModel.ONLINE);
+ ExternalView externalView = new ExternalView(REALTIME_TABLE_NAME);
+ externalView.setState(committingAllErrorSegment, server0,
SegmentStateModel.ERROR);
+ externalView.setState(committingAllErrorSegment, server1,
SegmentStateModel.ERROR);
+ segmentManager._idealState = idealState;
+
when(mockHelixResourceManager.getTableExternalView(REALTIME_TABLE_NAME)).thenReturn(externalView);
+ mockSegmentZKMetadataStatus(mockHelixResourceManager,
committingAllErrorSegment, Status.COMMITTING);
+
+ segmentManager.repairSegmentsInErrorState(tableConfig, true);
+
+ // With the override, the re-ingestion path is taken (server lookup is
attempted) and no reset is issued
+ verify(mockHelixResourceManager).getDataInstanceAdminEndpoints(any());
Review Comment:
Right: the unstubbed `BiMap` came back null, and the NPE was swallowed. In
a779ff97dd:
- `triggerReingestion` is package-private (`@VisibleForTesting`), and the
test fake records its calls.
- The override test stubs `getDataInstanceAdminEndpoints` with a populated
`BiMap`. It asserts that re-ingestion was triggered once, for the COMMITTING
segment, on an endpoint of one of its servers.
- New tests cover the missing-endpoint and failed-trigger paths.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -3020,13 +3020,13 @@ public void repairSegmentsInErrorState(TableConfig
tableConfig,
if (status == Status.IN_PROGRESS &&
segmentsInErrorStateInAllReplicas.contains(segmentName)) {
LOGGER.info("Resetting IN_PROGRESS segment: {} in table: {} since all
replicas are in ERROR state.",
segmentName, realtimeTableName);
- _helixResourceManager.resetSegment(realtimeTableName, segmentName,
null);
+ resetErrorReplicas(realtimeTableName, segmentName, currentStateMap);
continue;
}
// 3) For all other non-IN_PROGRESS statuses with any ERROR replica,
trigger a reset to download/refresh
if (status != Status.IN_PROGRESS) {
- _helixResourceManager.resetSegment(realtimeTableName, segmentName,
null);
+ resetErrorReplicas(realtimeTableName, segmentName, currentStateMap);
Review Comment:
Partly agreed. I first argued this state was transient, and that doesn't
hold. For example, the other replica can be OFFLINE (its instance disabled)
while the commit is stuck, and then every reset waits out the download timeout
and fails again.
Skipping the reset isn't safe either, though. On OFFLINE -> ONLINE the
server first tries its local copy, and
`tryLoadExistingSegmentWithoutRegistering` loads a local COMMITTING copy
without a CRC check. That covers, for example, a committer that built the
segment and then failed. The controller can't tell whether such a copy exists,
so only the reset can find out.
So in a779ff97dd the reset still runs, as on master for pauseless tables
without dedup. A segment with no deep-store copy (COMMITTING, or committed with
a peer download URL) and no ONLINE or CONSUMING replica is now counted in
`PAUSELESS_SEGMENTS_IN_UNRECOVERABLE_ERROR_COUNT`, with a WARN, so a stuck one
is visible. The wait the reset causes is bounded by the table's download
timeout, and it ends as soon as a peer is ONLINE or the commit completes.
Covered by `testRepairSegmentsInErrorStateCountsSegmentsWithNoOtherSource`.
A per-segment attempt cap or backoff would need state that survives across
passes and controllers. If you think it's needed, I'd rather add it for all
pauseless tables in a separate 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]
