suvodeep-pyne commented on code in PR #19909:
URL: https://github.com/apache/pinot/pull/19909#discussion_r4339388662
##########
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:
Agreed. In a779ff97dd, failures stay isolated: one failing instance doesn't
stop the other resets. But every segment that fails to reset is counted in
`PAUSELESS_SEGMENTS_IN_UNRECOVERABLE_ERROR_COUNT`. That now includes a reset
message that isn't written: `resetPartitions` checks the results of
`setChildren`, which used to be ignored. Once every segment has been processed
and the gauge is written, the method throws an exception naming the failed
segments. So the failure reaches the periodic task's `PERIODIC_TASK_ERROR`
meter again. Covered by
`testRepairSegmentsInErrorStateCountsFailedRepairsAsUnrecoverable` and
`SegmentResetTest.testResetPartitionsFailsWhenResetMessageIsNotWritten`.
##########
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:
Fixed in a779ff97dd: both annotations have `alwaysRun = true`. Each teardown
step runs on its own, and later failures are attached as suppressed, as in
`PauselessRealtimeIngestionSegmentCommitFailureTest`.
--
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]
