xiangfu0 commented on code in PR #19909:
URL: https://github.com/apache/pinot/pull/19909#discussion_r4321116699
##########
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:
**Unbounded reset loop for stuck partial-ERROR COMMITTING segments, with
10-minute Helix thread stalls per cycle.** A COMMITTING pauseless segment with
some (not all) replicas in ERROR lands here and is reset on every validation
run with no attempt cap or backoff. On the server, OFFLINE->ONLINE for a
COMMITTING segment enters `RealtimeTableDataManager.downloadSegment`'s
busy-wait (default `DEFAULT_SEGMENT_DOWNLOAD_TIMEOUT_MS` = 10 min) holding the
segment lock on the Helix state-transition thread; if the commit is permanently
stuck (committing server died after COMMIT_START) and no replica is ONLINE to
peer-download from, it times out back to ERROR and the next run resets it
again, forever. Since re-ingestion stays gated by default on
dedup/partial-upsert tables, this state can be permanent — yet it is never
counted in `PAUSELESS_SEGMENTS_IN_UNRECOVERABLE_ERROR_COUNT` (only the
all-ERROR skip branch increments it). Suggest skipping (and counting as
unrecoverable) COMMITTING segm
ents here, or adding attempt tracking/backoff.
##########
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) {
Review Comment:
**Unhandled case: IN_PROGRESS segment with only SOME replicas in ERROR
matches no branch.** Case 1 needs COMMITTING, case 2 needs all-ERROR, and this
guard excludes IN_PROGRESS — so a consuming segment with one ERROR replica is
silently skipped every run, while the new gauges now count it and the header
doc says a lagging replica can recover. Concretely: one replica of a consuming
segment errors while the other keeps consuming (the integration test's own
injected failure, observed before commit); nothing resets it until the segment
commits (hours on time-based flush), `PAUSELESS_SEGMENTS_IN_ERROR_COUNT` stays
nonzero with UNRECOVERABLE at 0, and with `enforceConsumptionInOrder` the next
segment's consumer on that server stays gated. Since `resetErrorReplicas` now
touches only ERROR instances, the all-ERROR precondition on case 2 (which
existed to protect healthy replicas from the old untargeted reset) no longer
protects anything — cases 2 and 3 can collapse into "resetErrorRep
licas unless COMMITTING && all-ERROR". If skipping live IN_PROGRESS replicas
is intentional, please log and count that cell explicitly.
##########
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);
Review Comment:
**Potential silent data loss on dedup tables (newly enabled by this PR).**
This reset of an all-ERROR IN_PROGRESS segment re-consumes from the stream — it
does not "only reload data another replica or the deep store already holds" as
the case-1 comment claims. Nothing cleans dedup metadata on the ERROR
transition: `onBecomeOfflineFromError` is a no-op,
`BaseTableDataManager.doOffloadSegment` unregisters the manager *before*
`offload()` (and `releaseSegment` is not in a finally), and
`SegmentDataManager.offload()`'s CAS means a failed offload is never retried.
So a mid-consumption CONSUMING->OFFLINE failure inside
`MutableSegmentImpl.offload()` (e.g. dedup `removeSegment` rethrows) leaves
orphaned PKs registered. After this reset, the fresh consumer's rows hit
`checkRecordPresentOrUpdate`, which treats the orphaned entries as present, and
silently drops them -> permanent row loss on all replicas, where pre-PR the
segment stayed a visible ERROR. Suggest having the reset path offlo
ad/clean dedup (and upsert) metadata for the segment before re-consuming, or
gating this branch for dedup tables like re-ingestion is.
##########
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 =
allowRepairOfCommittingSegments(repairErrorSegmentsForPartialUpsertOrDedup,
tableConfig);
int segmentsInUnRecoverableState = 0;
- if (isPauselessTable && !repairCommittingSegments) {
- // We do not run reingestion for dedup and partial upsert tables in
pauseless as it can
- // lead to data inconsistencies
- _controllerMetrics.setOrUpdateTableGauge(realtimeTableName,
- ControllerGauge.PAUSELESS_SEGMENTS_IN_UNRECOVERABLE_ERROR_COUNT,
segmentsInErrorStateInAllReplicas.size());
- return;
- } else {
- LOGGER.info("Repairing error segments in table: {}.", realtimeTableName);
- }
+ // Count segments with any ERROR replica so a single lagging replica is
visible in the metric
if (isPauselessTable) {
_controllerMetrics.setOrUpdateTableGauge(realtimeTableName,
ControllerGauge.PAUSELESS_SEGMENTS_IN_ERROR_COUNT,
- segmentsInErrorStateInAllReplicas.size());
+ segmentsInErrorStateInAtLeastOneReplica.size());
} else {
_controllerMetrics.setOrUpdateTableGauge(realtimeTableName,
ControllerGauge.SEGMENTS_IN_ERROR_STATE,
Review Comment:
**`SEGMENTS_IN_ERROR_STATE` now has two writers with different units.**
`SegmentStatusChecker` (SegmentStatusChecker.java:575) writes the count of
(segment, instance) ERROR *pairs* for the same per-table gauge, while this now
writes the count of *segments* with >=1 ERROR replica. For a non-pauseless
table with `autoResetErrorSegmentsOnValidation` enabled and 1 segment in ERROR
on 3 replicas, the gauge flaps between 3 and 1 depending on which periodic task
ran last. The dual-writer conflict predates this PR, but it was latent (this
writer almost always wrote 0 since all-replica ERROR is rare); after this
change both writers are routinely nonzero with different values, breaking
threshold alerts. Suggest giving the gauge a single owner (SegmentStatusChecker
always runs) and dropping the write here, or using a separate gauge name for
the repair-side count.
##########
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.",
Review Comment:
**Reset can convert a visible ERROR into a silent ingestion stall when a
stale segment data manager is still registered.** The ERROR path never
offloads, so if the original failure happened after `registerSegment` in
`doAddConsumingSegment` (RealtimeTableDataManager.java:632; e.g.
`startConsumption`/thread-start failure, or a Helix transition timeout landing
post-registration), the dead manager stays registered. The reset's
OFFLINE->CONSUMING then hits the "already exists, skipping adding it as
CONSUMING segment" early return (RealtimeTableDataManager.java:593-598) and
*succeeds*: EV shows CONSUMING with no live consumer, this repair loop no
longer sees the segment (it only scans ERROR states), and with
`enforceConsumptionInOrder` all later segments of the partition on that server
block behind it. The ONLINE path recovers from a stale manager (lines 484-496);
the CONSUMING path has no equivalent. Suggest offloading a stale registered
manager in `doAddConsumingSegment` (mirroring t
he ONLINE path) as part of enabling these resets.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -3036,6 +3036,24 @@ public void repairSegmentsInErrorState(TableConfig
tableConfig,
}
}
+ /// Resets only the replicas of the segment that are in ERROR state in the
external view. Healthy replicas are left
+ /// untouched: resetting a healthy replica takes it through OFFLINE, which
on dedup / upsert tables briefly drops its
+ /// keys from the metadata manager while other replicas keep consuming, and
can make the only good copy of a
+ /// segment without a deep-store copy unavailable.
+ private void resetErrorReplicas(String realtimeTableName, String
segmentName, Map<String, String> currentStateMap) {
Review Comment:
**ZK read amplification on the repair path.** Each `resetSegment(table,
segment, instance)` call re-reads the full table IdealState and ExternalView
from ZK, plus per-instance liveInstance/currentState/pending-message reads in
`resetPartitionAllState` — and this now runs once per ERROR replica instead of
once per segment. After a server bounce leaves hundreds of segments x 2-3 ERROR
replicas, one validation pass performs O(segments x replicas) full IS+EV reads
on the controller periodic-task thread, loading ZK exactly when the cluster is
degraded. `resetSegments(table, instance, errorSegmentsOnly=true)` /
`resetPartitionAllState(instance, table, Set<segments>)` already show the
batched instance->segments pattern; suggest accumulating an
instance-to-segments map across the loop and resetting once per instance
(ideally reusing the IS/EV already fetched at the top of this method).
##########
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:
**Misleading WARN on every run, even when nothing is disallowed.**
`allowRepairOfCommittingSegments` is evaluated eagerly here, so every
validation pass over a dedup/partial-upsert table that has *any* ERROR replica
— e.g. one benign, resettable replica this PR happily repairs — logs
"Disallowing re-ingestion of errored COMMITTING segments ... because dedup is
enabled" although no COMMITTING segment exists and nothing was disallowed. That
trains operators to ignore the message before the real COMMITTING-skip case
occurs. The policy rationale is also now duplicated across three comment blocks
(method header, case-1 comment, `resetErrorReplicas` doc) that will drift.
Suggest evaluating the gate lazily inside case 1 (its skip-WARN already covers
the real event) and keeping the rationale in one place.
##########
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:
**This test never actually exercises re-ingestion.**
`getDataInstanceAdminEndpoints` is unstubbed, so the mock returns null; the
resulting NPE is swallowed inside `pickServerToReingest`'s catch, `aliveServer`
comes back null, and the production code takes the "No alive server found ...
skipping re-ingestion" path. The test therefore only pins that the endpoint
lookup happened and no reset was issued — a regression anywhere after the
lookup (e.g. `triggerReingestion` no longer called, wrong server picked) stays
green. Suggest stubbing `getDataInstanceAdminEndpoints` to return a populated
BiMap so the re-ingestion attempt is actually reached, and verifying it (e.g.
via the HTTP trigger failure log path or a spy).
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PauselessDedupErrorReplicaRepairTest.java:
##########
@@ -0,0 +1,219 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.integration.tests;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.commons.io.FileUtils;
+import org.apache.helix.model.ExternalView;
+import org.apache.pinot.controller.ControllerConf;
+import
org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
+import
org.apache.pinot.integration.tests.realtime.utils.FailureInjectingTableConfig;
+import
org.apache.pinot.integration.tests.realtime.utils.FailureInjectingTableDataManagerProvider;
+import org.apache.pinot.server.starter.helix.HelixInstanceDataManagerConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
+import
org.apache.pinot.spi.config.table.ingestion.ParallelSegmentConsumptionPolicy;
+import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.stream.StreamConfigProperties;
+import org.apache.pinot.spi.utils.CommonConstants;
+import
org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.SegmentStateModel;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+import org.apache.pinot.util.TestUtils;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertNotNull;
+
+
+/// Verifies that on a pauseless dedup table WITHOUT the disaster recovery
override, the segment-level validation
+/// repairs ERROR replicas whose segment another replica holds, by resetting
only those replicas so they reload from
+/// a peer or the deep store.
+///
+/// The consuming transition is made to fail on one server only, so every
affected segment keeps one healthy replica
+/// and one ERROR replica. Re-ingestion is never needed for this repair, so it
must run even though re-ingestion is
+/// disallowed for dedup tables. The dedup consume gate also blocks the
failing server's consumers while any of its
+/// replicas is in ERROR, so without this repair the lagging server never
catches up.
+public class PauselessDedupErrorReplicaRepairTest extends
BaseClusterIntegrationTest {
+ private static final int NUM_PARTITIONS = 2;
+ private static final long VALIDATION_RERUN_INTERVAL_MS = 10_000L;
+
+ @Override
+ protected String getAvroTarFileName() {
+ return "dedupPauselessIngestionTestData.tar.gz";
+ }
+
+ @Override
+ protected int getRealtimeSegmentFlushSize() {
+ return 2;
+ }
+
+ @Override
+ protected String getSchemaFileName() {
+ return "dedupIngestionTestSchema.schema";
+ }
+
+ @Override
+ protected long getCountStarResult() {
+ // Two distinct records are expected with pk values of 0, 1.
+ return 2;
+ }
+
+ @Override
+ protected String getPartitionColumn() {
+ return "id";
+ }
+
+ @Override
+ protected int getNumReplicas() {
+ return 2;
+ }
+
+ @Override
+ protected IngestionConfig getIngestionConfig() {
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ Map<String, String> streamConfigs = getStreamConfigs();
+
streamConfigs.put(StreamConfigProperties.PAUSELESS_SEGMENT_DOWNLOAD_TIMEOUT_SECONDS,
"10");
+ StreamIngestionConfig streamIngestionConfig = new
StreamIngestionConfig(List.of(streamConfigs));
+ streamIngestionConfig.setPauselessConsumptionEnabled(true);
+
streamIngestionConfig.setParallelSegmentConsumptionPolicy(ParallelSegmentConsumptionPolicy.ALLOW_DURING_BUILD_ONLY);
+ streamIngestionConfig.setEnforceConsumptionInOrder(true);
+ // Intentionally no disasterRecoveryMode: repair of COMMITTING segments
stays disallowed for this dedup table,
+ // which must not stop the reset of ERROR replicas that have a healthy
copy elsewhere.
+ ingestionConfig.setStreamIngestionConfig(streamIngestionConfig);
+ return ingestionConfig;
+ }
+
+ @Override
+ protected void overrideControllerConf(Map<String, Object> properties) {
+
properties.put(ControllerConf.ControllerPeriodicTasksConf.ENABLE_DEEP_STORE_RETRY_UPLOAD_LLC_SEGMENT,
true);
+ // Recovery is triggered explicitly by the test; keep the scheduled
validator from racing it.
+
properties.put(ControllerConf.ControllerPeriodicTasksConf.REALTIME_SEGMENT_VALIDATION_INITIAL_DELAY_IN_SECONDS,
+ 3600);
+ }
+
+ @Override
+ protected PinotConfiguration getServerConf(int serverId) {
+ PinotConfiguration serverConf = super.getServerConf(serverId);
+ serverConf.setProperty("pinot.server.instance.segment.store.uri", "file:"
+ _controllerConfig.getDataDir());
+ serverConf.setProperty("pinot.server.instance." +
HelixInstanceDataManagerConfig.UPLOAD_SEGMENT_TO_DEEP_STORE,
+ "true");
+ serverConf.setProperty("pinot.server.instance." +
CommonConstants.Server.TABLE_DATA_MANAGER_PROVIDER_CLASS,
+ FailureInjectingTableDataManagerProvider.class.getName());
+ if (serverId == 1) {
+ // Fail the consuming transition of the first segment of each partition
on this server only, so each affected
+ // segment has one ERROR replica here and one healthy replica on the
other server.
+ serverConf.setProperty(
+ "pinot.server.instance." +
FailureInjectingTableDataManagerProvider.FAILURE_CONFIG_KEY + "."
+ + getTableName(), new FailureInjectingTableConfig(false, true,
NUM_PARTITIONS).toJson());
+ }
+ return serverConf;
+ }
+
+ @BeforeClass
+ public void setUp()
+ throws Exception {
+ TestUtils.ensureDirectoriesExistAndEmpty(_tempDir, _segmentDir, _tarDir);
+ startZk();
+ startKafka();
+ startController();
+ startBroker();
+ startServers(2);
+
+ List<File> avroFiles = unpackAvroData(_tempDir);
+ pushAvroIntoKafka(avroFiles);
+
+ Schema schema = createSchema();
+ addSchema(schema);
+ TableConfig tableConfig = createDedupTableConfig(avroFiles.get(0),
getPartitionColumn(), NUM_PARTITIONS);
+ assertNotNull(tableConfig.getDedupConfig());
+ tableConfig.getValidationConfig().setRetentionTimeUnit("DAYS");
+ tableConfig.getValidationConfig().setRetentionTimeValue("100000");
+
tableConfig.getValidationConfig().setPeerSegmentDownloadScheme(CommonConstants.HTTP_PROTOCOL);
+ addTableConfig(tableConfig);
+ }
+
+ @Test
+ public void testErrorReplicasAreRepairedWithoutDisasterRecoveryOverride()
+ throws Exception {
+ String realtimeTableName =
TableNameBuilder.REALTIME.tableNameWithType(getTableName());
+
+ TestUtils.waitForCondition(aVoid ->
getNumErrorReplicasInEV(realtimeTableName) > 0, 600_000L,
+ "Injected failures did not put any replica in ERROR state");
+
+ // Drive the segment-level validation periodically (as it runs in
production) until every ERROR replica is
+ // repaired. The runs are spaced out so a pass does not re-trigger resets
still in flight from the previous pass.
+ long[] lastValidationRunMs = new long[1];
+ TestUtils.waitForCondition(aVoid -> {
+ if (getNumErrorReplicasInEV(realtimeTableName) == 0) {
+ return true;
+ }
+ long nowMs = System.currentTimeMillis();
+ if (nowMs - lastValidationRunMs[0] >= VALIDATION_RERUN_INTERVAL_MS) {
+ lastValidationRunMs[0] = nowMs;
+ runSegmentLevelValidation();
+ }
+ return false;
+ }, 1000L, 600_000L, "ERROR replicas were not repaired without the disaster
recovery override");
+
+ // With all its replicas repaired, the lagging server passes the dedup
consume gate and catches up, so queries
+ // return the deduplicated count from either replica group.
+ waitForAllDocsLoaded(600_000L);
+ }
+
+ private int getNumErrorReplicasInEV(String realtimeTableName) {
+ ExternalView externalView = _helixResourceManager.getHelixAdmin()
+ .getResourceExternalView(_helixResourceManager.getHelixClusterName(),
realtimeTableName);
+ if (externalView == null) {
+ return 0;
+ }
+ int numErrorReplicas = 0;
+ for (Map<String, String> instanceStateMap :
externalView.getRecord().getMapFields().values()) {
+ for (String state : instanceStateMap.values()) {
+ if (state.equals(SegmentStateModel.ERROR)) {
+ numErrorReplicas++;
+ }
+ }
+ }
+ return numErrorReplicas;
+ }
+
+ private void runSegmentLevelValidation() {
+ Properties periodicTaskProperties = new Properties();
+
periodicTaskProperties.setProperty(ControllerPeriodicTask.RUN_SEGMENT_LEVEL_VALIDATION,
Boolean.TRUE.toString());
+
_controllerStarter.getRealtimeSegmentValidationManager().run(periodicTaskProperties);
+ }
+
+ @AfterClass
+ public void tearDown()
Review Comment:
**Fragile test lifecycle: one failure leaks the whole cluster into the
shared surefire fork.** Two issues: (1) `@BeforeClass`/`@AfterClass` lack
`alwaysRun = true` (the sibling
`PauselessRealtimeIngestionSegmentCommitFailureTest` has it), so if `setUp`
throws partway (Kafka/server start failure) TestNG skips this teardown
entirely; (2) even when it runs, this is an unguarded sequential chain — if
`dropRealtimeTable` throws (likely when the table is wedged with ERROR replicas
after a test failure), every subsequent `stop*` call and the temp-dir cleanup
are skipped, cascading port/state failures into later integration suites in the
same JVM. Suggest `alwaysRun = true` on both annotations and wrapping each
teardown step in try/catch as the sibling test does.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -3036,6 +3036,24 @@ public void repairSegmentsInErrorState(TableConfig
tableConfig,
}
}
+ /// Resets only the replicas of the segment that are in ERROR state in the
external view. Healthy replicas are left
+ /// untouched: resetting a healthy replica takes it through OFFLINE, which
on dedup / upsert tables briefly drops its
+ /// keys from the metadata manager while other replicas keep consuming, and
can make the only good copy of a
+ /// segment without a deep-store copy unavailable.
+ private void resetErrorReplicas(String realtimeTableName, String
segmentName, Map<String, String> currentStateMap) {
+ for (Map.Entry<String, String> entry : currentStateMap.entrySet()) {
+ if (SegmentStateModel.ERROR.equals(entry.getValue())) {
+ String instanceName = entry.getKey();
+ try {
Review Comment:
**Silent no-op when the EV instance is no longer in IdealState.**
`resetSegment(table, segment, instance)` goes through `parseInstanceSet`, which
returns `Set.of()` without any log or exception when the target instance is not
in the segment's IdealState instance set. After a rebalance moves a segment off
server A while A holds a stale ERROR replica in the EV, every validation pass
logs "Found N segments ... repairing them", then no-ops invisibly here (this
catch never fires), and the error gauges stay elevated indefinitely with no
actionable log line. Suggest filtering against the already-held
`segmentToInstanceIdealStateMap` before calling reset, or logging when the
instance is skipped.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -2983,14 +2977,20 @@ public void repairSegmentsInErrorState(TableConfig
tableConfig,
}
Status status = segmentZKMetadata.getStatus();
+ Map<String, String> currentStateMap =
segmentToInstanceCurrentStateMap.get(segmentName);
- // 1) For COMMITTING segments where all replicas are in ERROR, try
re-ingestion (subject to policy)
+ // 1) For COMMITTING segments where all replicas are in ERROR, try
re-ingestion (subject to policy). Re-ingestion
+ // is the only repair that can admit duplicate or inconsistent rows on
dedup / partial upsert tables, so it is
+ // the only repair gated on repairCommittingSegments. The resets below
only reload data another replica or the
+ // deep store already holds, so they stay enabled for such tables.
if (status == Status.COMMITTING &&
segmentsInErrorStateInAllReplicas.contains(segmentName)) {
if (!repairCommittingSegments) {
segmentsInUnRecoverableState += 1;
Review Comment:
**`PAUSELESS_SEGMENTS_IN_UNRECOVERABLE_ERROR_COUNT` is silently narrowed —
genuinely unrecoverable segments now read 0.** Pre-PR, the gated
dedup/partial-upsert path set this gauge to the count of all all-replica-ERROR
segments; now only this skip branch increments it. Cases that stay broken but
report 0: (a) a DONE segment with all replicas in ERROR and a peer:// download
URL whose deep-store upload never completed — every reset fails (no ONLINE
peer, no deep-store copy) and the replicas return to ERROR forever; (b) a
COMMITTING all-ERROR segment whose re-ingestion is attempted but finds no alive
server or whose trigger keeps failing (exception swallowed below); (c)
partial-ERROR stuck COMMITTING segments. The gauge is documented as "needs
manual intervention" (ControllerGauge.java:222-224), so alerts that fired
pre-PR go silent. Also note the gauge is now only written after the whole loop,
so an uncaught mid-loop exception (e.g. a ZK error from `getSegmentZKMetadata`,
not in
side any try/catch) leaves it stale for the run. Suggest counting
failed/unattemptable repairs as unrecoverable too.
##########
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 =
allowRepairOfCommittingSegments(repairErrorSegmentsForPartialUpsertOrDedup,
tableConfig);
int segmentsInUnRecoverableState = 0;
- if (isPauselessTable && !repairCommittingSegments) {
- // We do not run reingestion for dedup and partial upsert tables in
pauseless as it can
- // lead to data inconsistencies
- _controllerMetrics.setOrUpdateTableGauge(realtimeTableName,
- ControllerGauge.PAUSELESS_SEGMENTS_IN_UNRECOVERABLE_ERROR_COUNT,
segmentsInErrorStateInAllReplicas.size());
- return;
- } else {
- LOGGER.info("Repairing error segments in table: {}.", realtimeTableName);
- }
+ // Count segments with any ERROR replica so a single lagging replica is
visible in the metric
if (isPauselessTable) {
Review Comment:
**Breaking metric-semantics change under the same name.**
`PAUSELESS_SEGMENTS_IN_ERROR_COUNT` changes from "segments with ALL replicas in
ERROR" (a serving outage) to "segments with at least one ERROR replica". Alerts
calibrated to the old meaning now page on routine single-replica blips during
rolling restarts — the transient condition this PR auto-repairs moments later —
and conversely no gauge still exposes true full-outage segment counts
(`segmentsInErrorStateInAllReplicas` no longer feeds any gauge except the gated
COMMITTING skip). The comment in ControllerGauge.java was not updated either.
Suggest keeping this gauge's meaning and emitting a new, distinctly named gauge
(e.g. `SEGMENTS_WITH_ERROR_REPLICAS`) for repair visibility, or at minimum
calling the semantic change out in docs/release notes.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -3036,6 +3036,24 @@ public void repairSegmentsInErrorState(TableConfig
tableConfig,
}
}
+ /// Resets only the replicas of the segment that are in ERROR state in the
external view. Healthy replicas are left
+ /// untouched: resetting a healthy replica takes it through OFFLINE, which
on dedup / upsert tables briefly drops its
+ /// keys from the metadata manager while other replicas keep consuming, and
can make the only good copy of a
+ /// segment without a deep-store copy unavailable.
+ private void resetErrorReplicas(String realtimeTableName, String
segmentName, Map<String, String> currentStateMap) {
+ for (Map.Entry<String, String> entry : currentStateMap.entrySet()) {
+ if (SegmentStateModel.ERROR.equals(entry.getValue())) {
+ String instanceName = entry.getKey();
+ try {
+ _helixResourceManager.resetSegment(realtimeTableName, segmentName,
instanceName);
Review Comment:
**Race: can reset a replica that recovered to healthy, violating this
method's own doc.** The ERROR determination comes from the external-view
snapshot taken at the top of `repairSegmentsInErrorState`, but `resetSegment`
re-reads the EV and skips only OFFLINE instances
(`PinotHelixResourceManager.resetSegment`), and `resetPartitionAllState`
"resets all state regardless". The repair loop can be slow (per-segment ZK
reads, synchronous re-ingestion HTTP calls), so a replica that recovered
ERROR->ONLINE/CONSUMING in that window is bounced through OFFLINE anyway —
briefly dropping its dedup/upsert keys while serving, the exact harm the
javadoc above says this method avoids. Suggest re-checking that the instance is
still ERROR against the fresh EV inside the reset, as `resetSegments(table,
instance, errorSegmentsOnly=true)` already does.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -3036,6 +3036,24 @@ public void repairSegmentsInErrorState(TableConfig
tableConfig,
}
}
+ /// Resets only the replicas of the segment that are in ERROR state in the
external view. Healthy replicas are left
+ /// untouched: resetting a healthy replica takes it through OFFLINE, which
on dedup / upsert tables briefly drops its
+ /// keys from the metadata manager while other replicas keep consuming, and
can make the only good copy of a
+ /// segment without a deep-store copy unavailable.
+ private void resetErrorReplicas(String realtimeTableName, String
segmentName, Map<String, String> currentStateMap) {
+ for (Map.Entry<String, String> entry : currentStateMap.entrySet()) {
+ if (SegmentStateModel.ERROR.equals(entry.getValue())) {
+ String instanceName = entry.getKey();
+ try {
+ _helixResourceManager.resetSegment(realtimeTableName, segmentName,
instanceName);
+ } catch (Exception e) {
Review Comment:
**Swallowing reset failures hides persistent repair breakage.** The old
untargeted `resetSegment` call let failures (`RuntimeException("Reset segment
failed...")`, IllegalState preconditions) propagate out of
`repairSegmentsInErrorState` into the periodic task's error accounting. Now
every failure — including persistent ones like Helix message sends failing for
one server, or the segment being deleted between the IS snapshot and the reset
— is downgraded to an ERROR log, the run completes "successfully", and
`PAUSELESS_SEGMENTS_IN_UNRECOVERABLE_ERROR_COUNT` is written as 0. Monitoring
keyed on validation-task failures or that gauge misses the condition entirely.
Per-instance isolation is good, but consider counting failed resets (e.g. into
the unrecoverable gauge or a new failure meter) so repeated failures stay
visible.
--
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]
