This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 2877723da38 Repair ERROR replicas on pauseless dedup and
partial-upsert tables (#19909)
2877723da38 is described below
commit 2877723da384b783df372fb203fbf9fef0aea740
Author: Suvodeep Pyne <[email protected]>
AuthorDate: Wed Oct 21 16:17:49 2026 -0700
Repair ERROR replicas on pauseless dedup and partial-upsert tables (#19909)
---
.../helix/core/PinotHelixResourceManager.java | 105 +++++++-
.../realtime/PinotLLCRealtimeSegmentManager.java | 125 ++++++---
.../RealtimeSegmentValidationManager.java | 3 +-
.../controller/helix/core/SegmentResetTest.java | 201 +++++++++++++++
.../PinotLLCRealtimeSegmentManagerTest.java | 268 +++++++++++++++++++-
.../PauselessDedupErrorReplicaRepairTest.java | 281 +++++++++++++++++++++
.../realtime/utils/PauselessRealtimeTestUtils.java | 28 +-
7 files changed, 958 insertions(+), 53 deletions(-)
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
index 2d536a21b42..67793bec133 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
@@ -3357,6 +3357,55 @@ public class PinotHelixResourceManager {
}
}
+ /// Resets the replicas of the given segments that are in ERROR state,
leaving the other replicas untouched. The ideal
+ /// state and external view are read once, when the resets are sent, to find
the ERROR replicas; a replica on an
+ /// instance the ideal state no longer assigns the segment to is skipped.
The resets are sent in one batch per
+ /// instance, only for the segments still in ERROR in the instance's current
state, and the instance discards a reset
+ /// if the segment has left ERROR by the time it processes it.
+ ///
+ /// @return the segments whose reset failed on at least one instance
+ public Set<String> resetErrorReplicas(String tableNameWithType, Set<String>
segmentNames) {
+ IdealState idealState = getTableIdealState(tableNameWithType);
+ Preconditions.checkState(idealState != null, "Could not find ideal state
for table: %s", tableNameWithType);
+ ExternalView externalView = getTableExternalView(tableNameWithType);
+ Preconditions.checkState(externalView != null, "Could not find external
view for table: %s", tableNameWithType);
+
+ Map<String, Set<String>> instanceToResetSegmentsMap = new HashMap<>();
+ for (String segmentName : segmentNames) {
+ Map<String, String> externalViewStateMap =
externalView.getStateMap(segmentName);
+ Map<String, String> idealStateMap =
idealState.getInstanceStateMap(segmentName);
+ if (externalViewStateMap == null || idealStateMap == null) {
+ continue;
+ }
+ for (Map.Entry<String, String> entry : externalViewStateMap.entrySet()) {
+ String instance = entry.getKey();
+ if (!SegmentStateModel.ERROR.equals(entry.getValue())) {
+ continue;
+ }
+ if (idealStateMap.containsKey(instance)) {
+ instanceToResetSegmentsMap.computeIfAbsent(instance, i -> new
HashSet<>()).add(segmentName);
+ } else {
+ LOGGER.warn("Skipping reset of ERROR segment: {} of table: {} on
instance: {} which the ideal state no "
+ + "longer assigns it to", segmentName, tableNameWithType,
instance);
+ }
+ }
+ }
+
+ Set<String> failedSegments = new HashSet<>();
+ for (Map.Entry<String, Set<String>> entry :
instanceToResetSegmentsMap.entrySet()) {
+ String instance = entry.getKey();
+ Set<String> resetSegments = entry.getValue();
+ try {
+ resetPartitions(instance, tableNameWithType, resetSegments, true);
+ } catch (Exception e) {
+ LOGGER.error("Failed to reset ERROR segments: {} of table: {} on
instance: {}", resetSegments,
+ tableNameWithType, instance, e);
+ failedSegments.addAll(resetSegments);
+ }
+ }
+ return failedSegments;
+ }
+
private void resetPartitionAllState(String instance, String
tableNameWithType, Set<String> segmentNames,
List<String> failedInstances) {
try {
@@ -3384,9 +3433,22 @@ public class PinotHelixResourceManager {
/// However instead of resetting only the ERROR state to its initial state.
we reset all state regardless.
@VisibleForTesting
void resetPartitionAllState(String instanceName, String resourceName,
Set<String> resetPartitionNames) {
+ resetPartitions(instanceName, resourceName, resetPartitionNames, false);
+ }
+
+ /// Resets the given partitions on the instance to the initial state.
+ ///
+ /// With `errorStateOnly`, like [HelixAdmin#resetPartition(String, String,
String, List)], only the partitions in
+ /// ERROR state in the instance's current state are reset, and each message
is sent from ERROR, so the instance
+ /// discards it if the partition has left ERROR by the time it processes it.
A partition with a pending state
+ /// transition message is skipped. Otherwise every partition is reset from
its current state, and a pending state
+ /// transition message for any of them fails the call.
+ @VisibleForTesting
+ void resetPartitions(String instanceName, String resourceName, Set<String>
resetPartitionNames,
+ boolean errorStateOnly) {
LOGGER.info("Resetting partitions: {} for resource: {} on instance: {}",
resetPartitionNames, resourceName,
instanceName);
- HelixDataAccessor accessor = _helixZkManager.getHelixDataAccessor();
+ HelixDataAccessor accessor = getHelixZkManager().getHelixDataAccessor();
PropertyKey.Builder keyBuilder = accessor.keyBuilder();
// check the instance is alive
@@ -3409,6 +3471,7 @@ public class PinotHelixResourceManager {
CurrentState curState =
accessor.getProperty(keyBuilder.currentState(instanceName, sessionId,
resourceName));
// check there is no pending messages for the partitions exist
+ Set<String> partitionsWithPendingMessage = new HashSet<>();
List<Message> messages =
accessor.getChildValues(keyBuilder.messages(instanceName), true);
for (Message message : messages) {
if
(!Message.MessageType.STATE_TRANSITION.name().equalsIgnoreCase(message.getMsgType())
|| !sessionId.equals(
@@ -3416,9 +3479,12 @@ public class PinotHelixResourceManager {
|| !resetPartitionNames.contains(message.getPartitionName())) {
continue;
}
- throw new RuntimeException(
- "Can't reset state for " + resourceName + "." + resetPartitionNames
+ " on " + instanceName
- + ", because a pending message " + message + " exists for
resource " + message.getResourceName());
+ if (!errorStateOnly) {
+ throw new RuntimeException(
+ "Can't reset state for " + resourceName + "." +
resetPartitionNames + " on " + instanceName
+ + ", because a pending message " + message + " exists for
resource " + message.getResourceName());
+ }
+ partitionsWithPendingMessage.add(message.getPartitionName());
}
String adminName;
@@ -3433,6 +3499,19 @@ public class PinotHelixResourceManager {
List<Message> resetMessages = new ArrayList<>();
List<PropertyKey> messageKeys = new ArrayList<>();
for (String partitionName : resetPartitionNames) {
+ String fromState = curState.getState(partitionName);
+ if (errorStateOnly) {
+ if (!SegmentStateModel.ERROR.equals(fromState)) {
+ LOGGER.info("Skipping reset of partition: {} for resource: {} on
instance: {} in state: {}", partitionName,
+ resourceName, instanceName, fromState);
+ continue;
+ }
+ if (partitionsWithPendingMessage.contains(partitionName)) {
+ LOGGER.info("Skipping reset of partition: {} for resource: {} on
instance: {} with a pending message",
+ partitionName, resourceName, instanceName);
+ continue;
+ }
+ }
// send currentState to initialState message
String msgId = UUID.randomUUID().toString();
Message message = new Message(Message.MessageType.STATE_TRANSITION,
msgId);
@@ -3443,7 +3522,7 @@ public class PinotHelixResourceManager {
message.setResourceName(resourceName);
message.setTgtSessionId(sessionId);
message.setStateModelDef(stateModelDef);
- message.setFromState(curState.getState(partitionName));
+ message.setFromState(fromState);
message.setToState(stateModel.getInitialState());
message.setStateModelFactoryName(idealState.getStateModelFactoryName());
@@ -3458,7 +3537,21 @@ public class PinotHelixResourceManager {
messageKeys.add(keyBuilder.message(instanceName, message.getId()));
}
- accessor.setChildren(messageKeys, resetMessages);
+ if (resetMessages.isEmpty()) {
+ return;
+ }
+ boolean[] results = accessor.setChildren(messageKeys, resetMessages);
+ List<String> failedPartitions = new ArrayList<>();
+ for (int i = 0; i < results.length; i++) {
+ if (!results[i]) {
+ failedPartitions.add(resetMessages.get(i).getPartitionName());
+ }
+ }
+ if (!failedPartitions.isEmpty()) {
+ throw new RuntimeException(
+ "Failed to send reset messages for partitions: " + failedPartitions
+ " of resource: " + resourceName
+ + " on instance: " + instanceName);
+ }
}
/// Sends a segment refresh message to:
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
index d3211b55feb..56d8ce41dd3 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
@@ -22,6 +22,7 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Preconditions;
+import com.google.common.base.Suppliers;
import com.google.common.collect.BiMap;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
@@ -3103,7 +3104,22 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
///
/// API signature: POST http://\[serverURL\]/reingestSegment/\[segmentName\]
Request body (JSON):
///
- /// If segment is in ERROR state in only few replicas but has download URL,
we instead trigger a segment reset
+ /// Otherwise the ERROR replicas are reset:
+ /// - A segment still CONSUMING in the ideal state (IN_PROGRESS, or
COMMITTING before its commit updates the ideal
+ /// state) re-consumes when reset. It is reset only when all its replicas
are in ERROR; otherwise it is left alone
+ /// until it commits, so its ERROR replicas then reload it instead.
+ /// - Any other segment reloads when reset, from the replica's local copy,
the deep store or an ONLINE replica.
+ ///
+ /// Consuming from the stream again is unsafe on dedup and partial upsert
tables: re-ingestion can admit duplicate
+ /// or inconsistent rows, and a re-consumed segment can drop rows as
duplicates of keys left by the failed attempt,
+ /// or merge partial upsert rows against previous values that no replica has
loaded. So unless
+ /// [#allowRepairOfCommittingSegments] allows it, such tables skip
re-ingestion and, when pauseless, the reset that
+ /// re-consumes an IN_PROGRESS segment. Resets that reload a segment run for
every table.
+ ///
+ /// Segments counted as unrecoverable are those this pass cannot repair or
cannot issue a repair for: skipped by the
+ /// policy above, with no replica left consuming, with nothing but a
possible local copy to load from, with no server
+ /// to re-ingest on, or whose re-ingestion trigger or reset fails. A failed
reset fails the call once every segment
+ /// is processed.
///
/// @param tableConfig The table config
public void repairSegmentsInErrorState(TableConfig tableConfig,
@@ -3177,18 +3193,11 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
segmentsInErrorStateInAtLeastOneReplica.size(),
segmentsInErrorStateInAtLeastOneReplica,
segmentsInErrorStateInAllReplicas.size(),
segmentsInErrorStateInAllReplicas, realtimeTableName);
- 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);
- }
+ // Evaluated lazily, since it logs why repair is disallowed
+ Supplier<Boolean> repairCommittingSegments = Suppliers.memoize(
+ () ->
allowRepairOfCommittingSegments(repairErrorSegmentsForPartialUpsertOrDedup,
tableConfig));
+ Set<String> unrecoverableSegments = new HashSet<>();
+ Set<String> segmentsToReset = new HashSet<>();
if (isPauselessTable) {
_controllerMetrics.setOrUpdateTableGauge(realtimeTableName,
ControllerGauge.PAUSELESS_SEGMENTS_IN_ERROR_COUNT,
@@ -3207,14 +3216,19 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
}
Status status = segmentZKMetadata.getStatus();
+ boolean allReplicasInError =
segmentsInErrorStateInAllReplicas.contains(segmentName);
+ Map<String, String> idealStateMap =
segmentToInstanceIdealStateMap.get(segmentName);
+ Map<String, String> currentStateMap =
segmentToInstanceCurrentStateMap.get(segmentName);
// 1) For COMMITTING segments where all replicas are in ERROR, try
re-ingestion (subject to policy)
- if (status == Status.COMMITTING &&
segmentsInErrorStateInAllReplicas.contains(segmentName)) {
- if (!repairCommittingSegments) {
- segmentsInUnRecoverableState += 1;
- LOGGER.info(
- "Segment: {} in table: {} is COMMITTING with all replicas in
ERROR state. Skipping re-ingestion since "
- + "repairErrorSegments is false.",
+ if (status == Status.COMMITTING && allReplicasInError) {
+ if (!repairCommittingSegments.get()) {
+ unrecoverableSegments.add(segmentName);
+ LOGGER.warn(
+ "Segment: {} in table: {} is COMMITTING with all replicas in
ERROR state. Skipping re-ingestion because "
+ + "repair of committing segments is disabled for dedup and
partial upsert tables. Set "
+ + "disasterRecoveryMode=ALWAYS or run the validation task
with "
+ + "repairErrorSegmentsForPartialUpsertOrDedup=true to repair
it.",
segmentName, realtimeTableName);
continue;
}
@@ -3223,10 +3237,9 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
segmentName, realtimeTableName);
// Find at least one server that should host this segment and is alive
- Map<String, String> idealStateMap =
segmentToInstanceIdealStateMap.get(segmentName);
- assert idealStateMap != null;
String aliveServer = pickServerToReingest(idealStateMap.keySet());
if (aliveServer == null) {
+ unrecoverableSegments.add(segmentName);
LOGGER.warn("No alive server found to re-ingest segment: {} in
table: {}, skipping re-ingestion", segmentName,
realtimeTableName);
continue;
@@ -3235,28 +3248,70 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
triggerReingestion(aliveServer, segmentName);
LOGGER.info("Successfully triggered re-ingestion for segment: {} on
server: {}", segmentName, aliveServer);
} catch (Exception e) {
+ unrecoverableSegments.add(segmentName);
LOGGER.error("Failed to trigger re-ingestion for segment: {} on
server: {}", segmentName, aliveServer, e);
}
continue;
}
- // 2) For IN_PROGRESS segments where all replicas are in ERROR, issue a
reset to allow ERROR->OFFLINE and retry
- if (status == Status.IN_PROGRESS &&
segmentsInErrorStateInAllReplicas.contains(segmentName)) {
+ // 2) A reset re-consumes a segment that is still CONSUMING in the ideal
state: an IN_PROGRESS segment, or a
+ // COMMITTING one whose commit has not updated the ideal state yet
+ if (status == Status.IN_PROGRESS || (status == Status.COMMITTING &&
idealStateMap.containsValue(
+ SegmentStateModel.CONSUMING))) {
+ // With a replica not in ERROR, wait for the segment to commit, so its
ERROR replicas then reload it instead
+ if (!allReplicasInError) {
+ if (currentStateMap.containsValue(SegmentStateModel.CONSUMING)) {
+ LOGGER.info("Segment: {} in table: {} is still consuming with some
replicas in ERROR state. Leaving them "
+ + "to be reset once the segment commits.", segmentName,
realtimeTableName);
+ } else {
+ unrecoverableSegments.add(segmentName);
+ LOGGER.warn("Segment: {} in table: {} is still CONSUMING in the
ideal state, but no replica is consuming "
+ + "it and the others are in ERROR state. Leaving them to be
reset once the segment commits.",
+ segmentName, realtimeTableName);
+ }
+ continue;
+ }
+ if (isPauselessTable && !repairCommittingSegments.get()) {
+ unrecoverableSegments.add(segmentName);
+ LOGGER.warn(
+ "Segment: {} in table: {} is IN_PROGRESS with all replicas in
ERROR state. Skipping the reset, which "
+ + "re-consumes it, because repair by re-consuming is
disabled for pauseless dedup and partial upsert "
+ + "tables. Set disasterRecoveryMode=ALWAYS or run the
validation task with "
+ + "repairErrorSegmentsForPartialUpsertOrDedup=true to repair
it.",
+ segmentName, realtimeTableName);
+ continue;
+ }
LOGGER.info("Resetting IN_PROGRESS segment: {} in table: {} since all
replicas are in ERROR state.",
segmentName, realtimeTableName);
- _helixResourceManager.resetSegment(realtimeTableName, segmentName,
null);
+ segmentsToReset.add(segmentName);
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);
+ // 3) Otherwise a reset reloads the segment: from the replica's local
copy, the deep store, or an ONLINE replica.
+ // A COMMITTING segment, or one committed with a peer download URL, has
no deep-store copy. With no ONLINE or
+ // CONSUMING replica either, only a local copy can still load it, which
only the reset can tell.
+ segmentsToReset.add(segmentName);
+ boolean inDeepStore = status != Status.COMMITTING
+ &&
!CommonConstants.Segment.METADATA_URI_FOR_PEER_DOWNLOAD.equals(segmentZKMetadata.getDownloadUrl());
+ if (!inDeepStore &&
!currentStateMap.containsValue(SegmentStateModel.ONLINE)
+ && !currentStateMap.containsValue(SegmentStateModel.CONSUMING)) {
+ unrecoverableSegments.add(segmentName);
+ LOGGER.warn("Segment: {} in table: {} has no deep-store copy and no
ONLINE or CONSUMING replica. Resetting its "
+ + "ERROR replicas, which can only load a local copy.",
segmentName, realtimeTableName);
}
}
+ Set<String> segmentsFailedToReset =
+ segmentsToReset.isEmpty() ? Set.of() :
_helixResourceManager.resetErrorReplicas(realtimeTableName,
+ segmentsToReset);
+ unrecoverableSegments.addAll(segmentsFailedToReset);
if (isPauselessTable) {
_controllerMetrics.setOrUpdateTableGauge(realtimeTableName,
- ControllerGauge.PAUSELESS_SEGMENTS_IN_UNRECOVERABLE_ERROR_COUNT,
segmentsInUnRecoverableState);
+ ControllerGauge.PAUSELESS_SEGMENTS_IN_UNRECOVERABLE_ERROR_COUNT,
unrecoverableSegments.size());
+ }
+ if (!segmentsFailedToReset.isEmpty()) {
+ throw new RuntimeException(
+ "Failed to reset ERROR replicas of segments: " +
segmentsFailedToReset + " in table: " + realtimeTableName);
}
}
@@ -3264,9 +3319,10 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
return disasterRecoveryMode == DisasterRecoveryMode.ALWAYS;
}
- /// Whether to allow repairing the ERROR segments with ZK status: COMMITTING
- /// This method is only useful for pauseless ingestion with features like
dedup enabled (Since repairing committing
- /// segments for such tables can lead to incorrect data).
+ /// Whether to allow the repairs of ERROR segments that consume from the
stream again: re-ingesting a COMMITTING
+ /// segment, and on pauseless tables, resetting an IN_PROGRESS segment
+ /// This method is only useful for tables with features like dedup enabled
(Since such repairs can lead to incorrect
+ /// data for them).
/// @param repairErrorSegmentsForPartialUpsertOrDedup API context flag, if
true then always allow repair
/// @param tableConfig tableConfig
/// @return Returns true if repair is allowed
@@ -3290,7 +3346,7 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
== UpsertConfig.Mode.PARTIAL);
if (isPartialUpsertEnabled) {
// If isPartialUpsert is enabled, do not allow repair.
- LOGGER.warn("Skipping repair for errored segments in table: {} because
partialUpsert is enabled",
+ LOGGER.warn("Disallowing repairs that re-consume errored segments in
table: {} because partialUpsert is enabled",
tableConfig.getTableName());
return false;
}
@@ -3299,7 +3355,7 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
boolean isDedupEnabled = (dedupConfig != null) &&
(dedupConfig.isDedupEnabled());
if (isDedupEnabled) {
// If dedup is enabled, do not allow repair of error segment.
- LOGGER.warn("Skipping repair for errored segments in table: {} because
dedup is enabled",
+ LOGGER.warn("Disallowing repairs that re-consume errored segments in
table: {} because dedup is enabled",
tableConfig.getTableName());
return false;
}
@@ -3311,7 +3367,8 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
/// using Simple HTTP APIs.
///
/// POST http://\[serverURL\]/reingestSegment/\[segmentName\]
- private void triggerReingestion(String serverHostPort, String segmentName)
+ @VisibleForTesting
+ void triggerReingestion(String serverHostPort, String segmentName)
throws IOException, URISyntaxException, HttpErrorStatusException {
String scheme = CommonConstants.HTTP_PROTOCOL;
if (serverHostPort.contains(CommonConstants.HTTPS_PROTOCOL)) {
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
index e44706588db..fc7fdceed5a 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
@@ -134,7 +134,8 @@ public class RealtimeSegmentValidationManager extends
ControllerPeriodicTask<Rea
boolean isPauselessTable =
PauselessConsumptionUtils.isPauselessEnabled(tableConfig);
if (isPauselessTable || _segmentAutoResetOnErrorAtValidation) {
- // For realtime tables without dedup or partial upsert, repair segments
in error state.
+ // Repair segments in error state. Dedup and partial upsert tables skip
the repairs that consume from the stream
+ // again unless overridden (see repairSegmentsInErrorState).
// When pauseless consumption is disabled, this behavior remains gated by
// _segmentAutoResetOnErrorAtValidation to preserve legacy semantics.
_llcRealtimeSegmentManager.repairSegmentsInErrorState(tableConfig,
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/SegmentResetTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/SegmentResetTest.java
index 335255f5762..e15054a69a0 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/SegmentResetTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/SegmentResetTest.java
@@ -18,24 +18,43 @@
*/
package org.apache.pinot.controller.helix.core;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixManager;
+import org.apache.helix.PropertyKey;
+import org.apache.helix.model.CurrentState;
import org.apache.helix.model.ExternalView;
import org.apache.helix.model.IdealState;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.model.Message;
+import org.apache.helix.model.StateModelDefinition;
import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.spi.utils.CommonConstants;
+import
org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.SegmentStateModel;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.Test;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
public class SegmentResetTest {
+ private static final String SESSION_ID = "session_1";
private final String _tableName = "myTable_OFFLINE";
private final String _segmentName = "segment_1";
@@ -110,4 +129,186 @@ public class SegmentResetTest {
pinotHelixManager.resetSegment(_tableName, _segmentName, null);
pinotHelixManager.resetSegments(_tableName, null, false);
}
+
+ @Test
+ public void testResetErrorReplicasResetsOnlyReplicasStillInError() {
+ IdealState idealState = new IdealState(_tableName);
+ ExternalView externalView = new ExternalView(_tableName);
+ for (String segment : List.of("segment_1", "segment_2", "segment_3",
"segment_4")) {
+ idealState.setPartitionState(segment, _instance1,
SegmentStateModel.ONLINE);
+ idealState.setPartitionState(segment, _instance2,
SegmentStateModel.ONLINE);
+ }
+ externalView.setState("segment_1", _instance1, SegmentStateModel.ERROR);
+ externalView.setState("segment_1", _instance2, SegmentStateModel.ONLINE);
+ // An ERROR replica on an instance the ideal state no longer assigns the
segment to
+ externalView.setState("segment_1", "instance_3", SegmentStateModel.ERROR);
+ externalView.setState("segment_2", _instance1, SegmentStateModel.ERROR);
+ externalView.setState("segment_2", _instance2, SegmentStateModel.ERROR);
+ // Recovered after the caller saw it in ERROR
+ externalView.setState("segment_3", _instance1, SegmentStateModel.ONLINE);
+ externalView.setState("segment_3", _instance2, SegmentStateModel.ONLINE);
+ // In ERROR but not requested
+ externalView.setState("segment_4", _instance1, SegmentStateModel.ERROR);
+ ErrorReplicaResetHelixResourceManager pinotHelixManager =
+ new ErrorReplicaResetHelixResourceManager(idealState, externalView);
+
+ // segment_5 is in neither the ideal state nor the external view, e.g.
deleted after the caller looked
+ Set<String> failedSegments =
pinotHelixManager.resetErrorReplicas(_tableName,
+ Set.of("segment_1", "segment_2", "segment_3", "segment_5"));
+
+ assertEquals(failedSegments, Set.of());
+ assertEquals(pinotHelixManager._resetSegmentsPerInstance,
+ Map.of(_instance1, Set.of("segment_1", "segment_2"), _instance2,
Set.of("segment_2")));
+ }
+
+ @Test
+ public void testResetErrorReplicasReturnsSegmentsOfFailedInstance() {
+ IdealState idealState = new IdealState(_tableName);
+ ExternalView externalView = new ExternalView(_tableName);
+ for (String segment : List.of("segment_1", "segment_2")) {
+ idealState.setPartitionState(segment, _instance1,
SegmentStateModel.ONLINE);
+ idealState.setPartitionState(segment, _instance2,
SegmentStateModel.ONLINE);
+ }
+ externalView.setState("segment_1", _instance1, SegmentStateModel.ERROR);
+ externalView.setState("segment_1", _instance2, SegmentStateModel.ERROR);
+ externalView.setState("segment_2", _instance1, SegmentStateModel.ONLINE);
+ externalView.setState("segment_2", _instance2, SegmentStateModel.ERROR);
+ ErrorReplicaResetHelixResourceManager pinotHelixManager =
+ new ErrorReplicaResetHelixResourceManager(idealState, externalView);
+ pinotHelixManager._instancesToFail.add(_instance1);
+
+ Set<String> failedSegments =
pinotHelixManager.resetErrorReplicas(_tableName, Set.of("segment_1",
"segment_2"));
+
+ // The other instance is still reset
+ assertEquals(failedSegments, Set.of("segment_1"));
+ assertEquals(pinotHelixManager._resetSegmentsPerInstance,
+ Map.of(_instance2, Set.of("segment_1", "segment_2")));
+ }
+
+ /// Serves the given ideal state and external view, and records the resets
per instance instead of sending them.
+ private static class ErrorReplicaResetHelixResourceManager extends
PinotHelixResourceManager {
+ private final IdealState _idealState;
+ private final ExternalView _externalView;
+ private final Set<String> _instancesToFail = new HashSet<>();
+ private final Map<String, Set<String>> _resetSegmentsPerInstance = new
HashMap<>();
+
+ ErrorReplicaResetHelixResourceManager(IdealState idealState, ExternalView
externalView) {
+ super(createControllerConf());
+ _idealState = idealState;
+ _externalView = externalView;
+ }
+
+ private static ControllerConf createControllerConf() {
+ ControllerConf controllerConf = new ControllerConf();
+ controllerConf.setZkStr("localhost:2181");
+ controllerConf.setHelixClusterName("cluster01");
+ return controllerConf;
+ }
+
+ @Override
+ public IdealState getTableIdealState(String tableNameWithType) {
+ return _idealState;
+ }
+
+ @Override
+ public ExternalView getTableExternalView(String tableNameWithType) {
+ return _externalView;
+ }
+
+ @Override
+ void resetPartitions(String instanceName, String resourceName, Set<String>
resetPartitionNames,
+ boolean errorStateOnly) {
+ assertTrue(errorStateOnly);
+ if (_instancesToFail.contains(instanceName)) {
+ throw new RuntimeException("Test: Fail reset for " + instanceName);
+ }
+ _resetSegmentsPerInstance.put(instanceName, resetPartitionNames);
+ }
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void
testResetPartitionsErrorStateOnlySendsResetsFromErrorForPartitionsInError() {
+ CurrentState currentState = new CurrentState(_tableName);
+ currentState.setState("segment_1", SegmentStateModel.ERROR);
+ // Recovered after the external view was read
+ currentState.setState("segment_2", SegmentStateModel.ONLINE);
+ // A reset from an earlier pass is still pending
+ currentState.setState("segment_3", SegmentStateModel.ERROR);
+ Message pendingMessage = new Message(Message.MessageType.STATE_TRANSITION,
"pending");
+ pendingMessage.setTgtSessionId(SESSION_ID);
+ pendingMessage.setResourceName(_tableName);
+ pendingMessage.setPartitionName("segment_3");
+ HelixDataAccessor accessor = mockHelixDataAccessor(currentState,
List.of(pendingMessage));
+ when(accessor.setChildren(anyList(), anyList())).thenReturn(new
boolean[]{true});
+
+ new HelixDataAccessorResourceManager(accessor).resetPartitions(_instance1,
_tableName,
+ Set.of("segment_1", "segment_2", "segment_3"), true);
+
+ ArgumentCaptor<List<Message>> messagesCaptor =
ArgumentCaptor.forClass(List.class);
+ verify(accessor).setChildren(anyList(), messagesCaptor.capture());
+ List<Message> messages = messagesCaptor.getValue();
+ assertEquals(messages.size(), 1);
+ assertEquals(messages.get(0).getPartitionName(), "segment_1");
+ // Sent from ERROR, so the instance discards it if the segment leaves
ERROR before processing it
+ assertEquals(messages.get(0).getFromState(), SegmentStateModel.ERROR);
+ assertEquals(messages.get(0).getToState(), SegmentStateModel.OFFLINE);
+ }
+
+ @Test
+ public void testResetPartitionsFailsWhenResetMessageIsNotWritten() {
+ CurrentState currentState = new CurrentState(_tableName);
+ currentState.setState("segment_1", SegmentStateModel.ERROR);
+ HelixDataAccessor accessor = mockHelixDataAccessor(currentState,
List.of());
+ when(accessor.setChildren(anyList(), anyList())).thenReturn(new
boolean[]{false});
+
+ RuntimeException exception = expectThrows(RuntimeException.class,
+ () -> new
HelixDataAccessorResourceManager(accessor).resetPartitions(_instance1,
_tableName,
+ Set.of("segment_1"), true));
+ assertTrue(exception.getMessage().contains("segment_1"),
exception.getMessage());
+ }
+
+ private HelixDataAccessor mockHelixDataAccessor(CurrentState currentState,
List<Message> messages) {
+ LiveInstance liveInstance = mock(LiveInstance.class);
+ when(liveInstance.getEphemeralOwner()).thenReturn(SESSION_ID);
+ IdealState idealState = new IdealState(_tableName);
+ idealState.setStateModelDefRef("SegmentOnlineOfflineStateModel");
+ StateModelDefinition stateModelDefinition =
mock(StateModelDefinition.class);
+
when(stateModelDefinition.getInitialState()).thenReturn(SegmentStateModel.OFFLINE);
+ HelixDataAccessor accessor = mock(HelixDataAccessor.class);
+ when(accessor.keyBuilder()).thenReturn(new
PropertyKey.Builder("cluster01"));
+ when(accessor.getProperty(any(PropertyKey.class))).thenAnswer(invocation
-> {
+ PropertyKey key = invocation.getArgument(0);
+ switch (key.getType()) {
+ case LIVEINSTANCES:
+ return liveInstance;
+ case IDEALSTATES:
+ return idealState;
+ case STATEMODELDEFS:
+ return stateModelDefinition;
+ case CURRENTSTATES:
+ return currentState;
+ default:
+ throw new IllegalArgumentException("Unexpected property key: " +
key);
+ }
+ });
+ when(accessor.getChildValues(any(PropertyKey.class),
anyBoolean())).thenReturn(new ArrayList<>(messages));
+ return accessor;
+ }
+
+ /// Sends the resets through the given Helix data accessor.
+ private static class HelixDataAccessorResourceManager extends
PinotHelixResourceManager {
+ private final HelixManager _helixManager;
+
+ HelixDataAccessorResourceManager(HelixDataAccessor accessor) {
+ super(ErrorReplicaResetHelixResourceManager.createControllerConf());
+ _helixManager = mock(HelixManager.class);
+ when(_helixManager.getHelixDataAccessor()).thenReturn(accessor);
+ }
+
+ @Override
+ public HelixManager getHelixZkManager() {
+ return _helixManager;
+ }
+ }
}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java
index 4eeef13372e..9216dddf860 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java
@@ -19,6 +19,8 @@
package org.apache.pinot.controller.helix.core.realtime;
import com.google.common.base.Preconditions;
+import com.google.common.collect.BiMap;
+import com.google.common.collect.HashBiMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import java.io.File;
@@ -62,9 +64,11 @@ import org.apache.helix.store.zk.ZkHelixPropertyStore;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.pinot.common.assignment.InstancePartitions;
import org.apache.pinot.common.exception.HttpErrorStatusException;
+import org.apache.pinot.common.exception.InvalidConfigException;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metadata.segment.SegmentPartitionMetadata;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.metrics.ControllerGauge;
import org.apache.pinot.common.metrics.ControllerMeter;
import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.common.restlet.resources.BatchConfig;
@@ -85,12 +89,14 @@ import
org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
import org.apache.pinot.segment.spi.partition.metadata.ColumnPartitionMetadata;
import org.apache.pinot.spi.auth.AuthProvider;
import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
+import org.apache.pinot.spi.config.table.DedupConfig;
import org.apache.pinot.spi.config.table.DisasterRecoveryMode;
import org.apache.pinot.spi.config.table.PauseState;
import org.apache.pinot.spi.config.table.SegmentPartitionConfig;
import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.config.table.UpsertConfig;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
@@ -2896,6 +2902,249 @@ public class PinotLLCRealtimeSegmentManagerTest {
pinotLLCRealtimeSegmentManager.shouldRepairErrorSegmentsForPartialUpsertOrDedup(DisasterRecoveryMode.ALWAYS));
}
+ @Test
+ public void
testRepairSegmentsInErrorStateResetsCommittedSegmentsOnPauselessDedupTable()
+ throws InvalidConfigException {
+ PinotHelixResourceManager mockHelixResourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics mockControllerMetrics = mock(ControllerMetrics.class);
+ FakePinotLLCRealtimeSegmentManager segmentManager =
+ new FakePinotLLCRealtimeSegmentManager(mockHelixResourceManager,
mockControllerMetrics);
+ IdealState idealState = new IdealState(REALTIME_TABLE_NAME);
+ ExternalView externalView = new ExternalView(REALTIME_TABLE_NAME);
+ String committingPartialErrorSegment =
addSegmentForErrorRepair(mockHelixResourceManager, idealState,
+ externalView, 0, 0, Status.COMMITTING, SegmentStateModel.ONLINE,
SegmentStateModel.ERROR);
+ addSegmentForErrorRepair(mockHelixResourceManager, idealState,
externalView, 0, 1, Status.COMMITTING,
+ SegmentStateModel.ERROR, SegmentStateModel.ERROR);
+ // The next segment of the partition whose COMMITTING segment stays in
ERROR
+ addSegmentForErrorRepair(mockHelixResourceManager, idealState,
externalView, 0, 2, Status.IN_PROGRESS,
+ SegmentStateModel.ERROR, SegmentStateModel.ERROR);
+ String donePartialErrorSegment =
addSegmentForErrorRepair(mockHelixResourceManager, idealState, externalView, 1,
+ 0, Status.DONE, SegmentStateModel.ONLINE, SegmentStateModel.ERROR);
+ String doneAllErrorSegment =
addSegmentForErrorRepair(mockHelixResourceManager, idealState, externalView, 1,
1,
+ Status.DONE, SegmentStateModel.ERROR, SegmentStateModel.ERROR);
+ addSegmentForErrorRepair(mockHelixResourceManager, idealState,
externalView, 1, 2, Status.IN_PROGRESS,
+ SegmentStateModel.CONSUMING, SegmentStateModel.ERROR);
+ segmentManager._idealState = idealState;
+
when(mockHelixResourceManager.getTableExternalView(REALTIME_TABLE_NAME)).thenReturn(externalView);
+
+ segmentManager.repairSegmentsInErrorState(createPauselessTableConfig(new
DedupConfig(), null), false);
+
+ // Only the committed segments are reset: their ERROR replicas reload from
a peer or the deep store. Resetting an
+ // IN_PROGRESS segment would re-consume it, and the COMMITTING segment
with every replica in ERROR needs
+ // re-ingestion, both of which stay disallowed on a dedup table without
the override.
+ verify(mockHelixResourceManager).resetErrorReplicas(REALTIME_TABLE_NAME,
+ Set.of(committingPartialErrorSegment, donePartialErrorSegment,
doneAllErrorSegment));
+ verify(mockHelixResourceManager, never()).resetSegment(any(), any(),
any());
+ verify(mockHelixResourceManager,
never()).getDataInstanceAdminEndpoints(any());
+ // The error count keeps counting segments with every replica in ERROR.
The two left unrepaired are unrecoverable;
+ // the IN_PROGRESS segment with a healthy replica is not, since its ERROR
replica is reset once it commits.
+ verify(mockControllerMetrics).setOrUpdateTableGauge(REALTIME_TABLE_NAME,
+ ControllerGauge.PAUSELESS_SEGMENTS_IN_ERROR_COUNT, 3L);
+ verify(mockControllerMetrics).setOrUpdateTableGauge(REALTIME_TABLE_NAME,
+ ControllerGauge.PAUSELESS_SEGMENTS_IN_UNRECOVERABLE_ERROR_COUNT, 2L);
+ }
+
+ @Test
+ public void
testRepairSegmentsInErrorStateDoesNotReconsumeAfterErrorSegmentOnPauselessPartialUpsertTable()
+ throws InvalidConfigException {
+ PinotHelixResourceManager mockHelixResourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics mockControllerMetrics = mock(ControllerMetrics.class);
+ FakePinotLLCRealtimeSegmentManager segmentManager =
+ new FakePinotLLCRealtimeSegmentManager(mockHelixResourceManager,
mockControllerMetrics);
+ IdealState idealState = new IdealState(REALTIME_TABLE_NAME);
+ ExternalView externalView = new ExternalView(REALTIME_TABLE_NAME);
+ // Re-consuming the IN_PROGRESS segment would merge its rows against the
previous values held by the COMMITTING
+ // segment before it, which no replica has loaded
+ addSegmentForErrorRepair(mockHelixResourceManager, idealState,
externalView, 0, 0, Status.COMMITTING,
+ SegmentStateModel.ERROR, SegmentStateModel.ERROR);
+ addSegmentForErrorRepair(mockHelixResourceManager, idealState,
externalView, 0, 1, Status.IN_PROGRESS,
+ SegmentStateModel.ERROR, SegmentStateModel.ERROR);
+ segmentManager._idealState = idealState;
+
when(mockHelixResourceManager.getTableExternalView(REALTIME_TABLE_NAME)).thenReturn(externalView);
+
+ segmentManager.repairSegmentsInErrorState(
+ createPauselessTableConfig(null, new
UpsertConfig(UpsertConfig.Mode.PARTIAL)), false);
+
+ verify(mockHelixResourceManager, never()).resetErrorReplicas(any(), any());
+ verify(mockHelixResourceManager, never()).resetSegment(any(), any(),
any());
+ verify(mockHelixResourceManager,
never()).getDataInstanceAdminEndpoints(any());
+ verify(mockControllerMetrics).setOrUpdateTableGauge(REALTIME_TABLE_NAME,
+ ControllerGauge.PAUSELESS_SEGMENTS_IN_UNRECOVERABLE_ERROR_COUNT, 2L);
+ }
+
+ @Test
+ public void
testRepairSegmentsInErrorStateReingestsAndReconsumesOnPauselessDedupTableWithOverride()
+ throws InvalidConfigException {
+ PinotHelixResourceManager mockHelixResourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics mockControllerMetrics = mock(ControllerMetrics.class);
+ FakePinotLLCRealtimeSegmentManager segmentManager =
+ new FakePinotLLCRealtimeSegmentManager(mockHelixResourceManager,
mockControllerMetrics);
+ IdealState idealState = new IdealState(REALTIME_TABLE_NAME);
+ ExternalView externalView = new ExternalView(REALTIME_TABLE_NAME);
+ String committingAllErrorSegment =
addSegmentForErrorRepair(mockHelixResourceManager, idealState, externalView,
+ 0, 0, Status.COMMITTING, SegmentStateModel.ERROR,
SegmentStateModel.ERROR);
+ String inProgressAllErrorSegment =
addSegmentForErrorRepair(mockHelixResourceManager, idealState, externalView,
+ 0, 1, Status.IN_PROGRESS, SegmentStateModel.ERROR,
SegmentStateModel.ERROR);
+ segmentManager._idealState = idealState;
+
when(mockHelixResourceManager.getTableExternalView(REALTIME_TABLE_NAME)).thenReturn(externalView);
+ BiMap<String, String> serverEndpoints = HashBiMap.create(
+ Map.of(Helix.PREFIX_OF_SERVER_INSTANCE + 0, "http://server0:8097";,
Helix.PREFIX_OF_SERVER_INSTANCE + 1,
+ "http://server1:8097";));
+
when(mockHelixResourceManager.getDataInstanceAdminEndpoints(serverEndpoints.keySet())).thenReturn(
+ serverEndpoints);
+
+ segmentManager.repairSegmentsInErrorState(createPauselessTableConfig(new
DedupConfig(), null), true);
+
+ // With the override, the COMMITTING segment is re-ingested on one of the
servers hosting it, and the IN_PROGRESS
+ // segment is reset so it re-consumes
+ assertEquals(segmentManager._reingestionRequests.keySet(),
Set.of(committingAllErrorSegment));
+
assertTrue(serverEndpoints.containsValue(segmentManager._reingestionRequests.get(committingAllErrorSegment)));
+ verify(mockHelixResourceManager).resetErrorReplicas(REALTIME_TABLE_NAME,
Set.of(inProgressAllErrorSegment));
+ verify(mockControllerMetrics).setOrUpdateTableGauge(REALTIME_TABLE_NAME,
+ ControllerGauge.PAUSELESS_SEGMENTS_IN_UNRECOVERABLE_ERROR_COUNT, 0L);
+ }
+
+ @Test
+ public void
testRepairSegmentsInErrorStateCountsFailedRepairsAsUnrecoverable()
+ throws InvalidConfigException {
+ PinotHelixResourceManager mockHelixResourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics mockControllerMetrics = mock(ControllerMetrics.class);
+ FakePinotLLCRealtimeSegmentManager segmentManager =
+ new FakePinotLLCRealtimeSegmentManager(mockHelixResourceManager,
mockControllerMetrics);
+ IdealState idealState = new IdealState(REALTIME_TABLE_NAME);
+ ExternalView externalView = new ExternalView(REALTIME_TABLE_NAME);
+ addSegmentForErrorRepair(mockHelixResourceManager, idealState,
externalView, 0, 0, Status.COMMITTING,
+ SegmentStateModel.ERROR, SegmentStateModel.ERROR);
+ String donePartialErrorSegment =
addSegmentForErrorRepair(mockHelixResourceManager, idealState, externalView, 1,
+ 0, Status.DONE, SegmentStateModel.ONLINE, SegmentStateModel.ERROR);
+ segmentManager._idealState = idealState;
+
when(mockHelixResourceManager.getTableExternalView(REALTIME_TABLE_NAME)).thenReturn(externalView);
+
when(mockHelixResourceManager.getDataInstanceAdminEndpoints(any())).thenReturn(HashBiMap.create(
+ Map.of(Helix.PREFIX_OF_SERVER_INSTANCE + 0, "http://server0:8097";,
Helix.PREFIX_OF_SERVER_INSTANCE + 1,
+ "http://server1:8097";)));
+ segmentManager._failReingestion = true;
+ when(mockHelixResourceManager.resetErrorReplicas(REALTIME_TABLE_NAME,
Set.of(donePartialErrorSegment)))
+ .thenReturn(Set.of(donePartialErrorSegment));
+
+ // A failed reset fails the pass, as it did before resets were isolated
per segment, after the gauge is set
+ RuntimeException exception = expectThrows(RuntimeException.class,
+ () ->
segmentManager.repairSegmentsInErrorState(createPauselessTableConfig(null,
null), false));
+ assertTrue(exception.getMessage().contains(donePartialErrorSegment),
exception.getMessage());
+ verify(mockControllerMetrics).setOrUpdateTableGauge(REALTIME_TABLE_NAME,
+ ControllerGauge.PAUSELESS_SEGMENTS_IN_UNRECOVERABLE_ERROR_COUNT, 2L);
+ }
+
+ @Test
+ public void
testRepairSegmentsInErrorStateCountsCommittingSegmentWithoutServerEndpointAsUnrecoverable()
+ throws InvalidConfigException {
+ PinotHelixResourceManager mockHelixResourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics mockControllerMetrics = mock(ControllerMetrics.class);
+ FakePinotLLCRealtimeSegmentManager segmentManager =
+ new FakePinotLLCRealtimeSegmentManager(mockHelixResourceManager,
mockControllerMetrics);
+ IdealState idealState = new IdealState(REALTIME_TABLE_NAME);
+ ExternalView externalView = new ExternalView(REALTIME_TABLE_NAME);
+ addSegmentForErrorRepair(mockHelixResourceManager, idealState,
externalView, 0, 0, Status.COMMITTING,
+ SegmentStateModel.ERROR, SegmentStateModel.ERROR);
+ segmentManager._idealState = idealState;
+
when(mockHelixResourceManager.getTableExternalView(REALTIME_TABLE_NAME)).thenReturn(externalView);
+
when(mockHelixResourceManager.getDataInstanceAdminEndpoints(any())).thenReturn(HashBiMap.create());
+
+ segmentManager.repairSegmentsInErrorState(createPauselessTableConfig(null,
null), false);
+
+ assertTrue(segmentManager._reingestionRequests.isEmpty());
+ verify(mockControllerMetrics).setOrUpdateTableGauge(REALTIME_TABLE_NAME,
+ ControllerGauge.PAUSELESS_SEGMENTS_IN_UNRECOVERABLE_ERROR_COUNT, 1L);
+ }
+
+ @Test
+ public void testRepairSegmentsInErrorStateCountsSegmentsWithNoOtherSource()
+ throws InvalidConfigException {
+ PinotHelixResourceManager mockHelixResourceManager =
mock(PinotHelixResourceManager.class);
+ ControllerMetrics mockControllerMetrics = mock(ControllerMetrics.class);
+ FakePinotLLCRealtimeSegmentManager segmentManager =
+ new FakePinotLLCRealtimeSegmentManager(mockHelixResourceManager,
mockControllerMetrics);
+ IdealState idealState = new IdealState(REALTIME_TABLE_NAME);
+ ExternalView externalView = new ExternalView(REALTIME_TABLE_NAME);
+ // The commit has not updated the ideal state yet, so a reset would
re-consume these segments. They wait for the
+ // commit, which the second one has no replica left to finish.
+ String committingStillConsumingSegment =
addSegmentForErrorRepair(mockHelixResourceManager, idealState,
+ externalView, 0, 0, Status.COMMITTING, SegmentStateModel.CONSUMING,
SegmentStateModel.ERROR);
+ String committingStillConsumingWithoutConsumerSegment =
addSegmentForErrorRepair(mockHelixResourceManager,
+ idealState, externalView, 0, 1, Status.COMMITTING,
SegmentStateModel.OFFLINE, SegmentStateModel.ERROR);
+ for (String segmentName : List.of(committingStillConsumingSegment,
+ committingStillConsumingWithoutConsumerSegment)) {
+ idealState.setPartitionState(segmentName,
Helix.PREFIX_OF_SERVER_INSTANCE + 0, SegmentStateModel.CONSUMING);
+ idealState.setPartitionState(segmentName,
Helix.PREFIX_OF_SERVER_INSTANCE + 1, SegmentStateModel.CONSUMING);
+ }
+ // Not in the deep store yet: reset, since the ERROR replica may load a
local copy or wait for the other replica.
+ // With no ONLINE or CONSUMING replica, only a local copy is left.
+ String committingWithConsumingReplicaSegment =
addSegmentForErrorRepair(mockHelixResourceManager, idealState,
+ externalView, 1, 0, Status.COMMITTING, SegmentStateModel.CONSUMING,
SegmentStateModel.ERROR);
+ String committingWithOfflineReplicaSegment =
addSegmentForErrorRepair(mockHelixResourceManager, idealState,
+ externalView, 2, 0, Status.COMMITTING, SegmentStateModel.OFFLINE,
SegmentStateModel.ERROR);
+ // Committed with a peer download URL
+ String peerDownloadOnlineSegment =
addSegmentForErrorRepair(mockHelixResourceManager, idealState, externalView, 3,
+ 0, Status.DONE, SegmentStateModel.ONLINE, SegmentStateModel.ERROR);
+ String peerDownloadAllErrorSegment =
addSegmentForErrorRepair(mockHelixResourceManager, idealState, externalView,
+ 3, 1, Status.DONE, SegmentStateModel.ERROR, SegmentStateModel.ERROR);
+ for (String segmentName : List.of(peerDownloadOnlineSegment,
peerDownloadAllErrorSegment)) {
+ SegmentZKMetadata segmentZKMetadata = new SegmentZKMetadata(segmentName);
+ segmentZKMetadata.setStatus(Status.DONE);
+ segmentZKMetadata.setDownloadUrl(METADATA_URI_FOR_PEER_DOWNLOAD);
+ when(mockHelixResourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
segmentName)).thenReturn(
+ segmentZKMetadata);
+ }
+ segmentManager._idealState = idealState;
+
when(mockHelixResourceManager.getTableExternalView(REALTIME_TABLE_NAME)).thenReturn(externalView);
+ Set<String> segmentsToReset =
Set.of(committingWithConsumingReplicaSegment,
committingWithOfflineReplicaSegment,
+ peerDownloadOnlineSegment, peerDownloadAllErrorSegment);
+ // A segment whose reset also fails is counted once
+ when(mockHelixResourceManager.resetErrorReplicas(REALTIME_TABLE_NAME,
segmentsToReset)).thenReturn(
+ Set.of(peerDownloadAllErrorSegment));
+
+ expectThrows(RuntimeException.class,
+ () ->
segmentManager.repairSegmentsInErrorState(createPauselessTableConfig(null,
null), false));
+
+ verify(mockHelixResourceManager).resetErrorReplicas(REALTIME_TABLE_NAME,
segmentsToReset);
+ verify(mockControllerMetrics).setOrUpdateTableGauge(REALTIME_TABLE_NAME,
+ ControllerGauge.PAUSELESS_SEGMENTS_IN_UNRECOVERABLE_ERROR_COUNT, 3L);
+ }
+
+ private static TableConfig createPauselessTableConfig(@Nullable DedupConfig
dedupConfig,
+ @Nullable UpsertConfig upsertConfig) {
+ IngestionConfig ingestionConfig = new IngestionConfig();
+ StreamIngestionConfig streamIngestionConfig = new StreamIngestionConfig(
+
List.of(FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs().getStreamConfigsMap()));
+ streamIngestionConfig.setPauselessConsumptionEnabled(true);
+ ingestionConfig.setStreamIngestionConfig(streamIngestionConfig);
+ return new
TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME)
+ .setNumReplicas(2)
+ .setIngestionConfig(ingestionConfig)
+ .setDedupConfig(dedupConfig)
+ .setUpsertConfig(upsertConfig)
+ .build();
+ }
+
+ /// Adds a segment with one replica per given external view state, hosted on
servers 0, 1, ..., and mocks its ZK
+ /// metadata status. Returns the segment name.
+ private static String addSegmentForErrorRepair(PinotHelixResourceManager
mockHelixResourceManager,
+ IdealState idealState, ExternalView externalView, int partitionId, int
sequenceNumber, Status status,
+ String... replicaStates) {
+ String segmentName = new LLCSegmentName(RAW_TABLE_NAME, partitionId,
sequenceNumber, CURRENT_TIME_MS)
+ .getSegmentName();
+ String idealStateState = status == Status.IN_PROGRESS ?
SegmentStateModel.CONSUMING : SegmentStateModel.ONLINE;
+ for (int i = 0; i < replicaStates.length; i++) {
+ String server = Helix.PREFIX_OF_SERVER_INSTANCE + i;
+ idealState.setPartitionState(segmentName, server, idealStateState);
+ externalView.setState(segmentName, server, replicaStates[i]);
+ }
+ SegmentZKMetadata segmentZKMetadata = new SegmentZKMetadata(segmentName);
+ segmentZKMetadata.setStatus(status);
+ when(mockHelixResourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
segmentName)).thenReturn(
+ segmentZKMetadata);
+ return segmentName;
+ }
+
@Test
public void testOnChangeUpdatesMaxSegmentCompletionTime() {
FakePinotLLCRealtimeSegmentManager segmentManager = new
FakePinotLLCRealtimeSegmentManager();
@@ -3043,6 +3292,9 @@ public class PinotLLCRealtimeSegmentManagerTest {
};
FileUploadDownloadClient _mockedFileUploadDownloadClient;
PinotHelixResourceManager _mockResourceManager;
+ // Segment name to the server endpoint re-ingestion was triggered on
+ Map<String, String> _reingestionRequests = new HashMap<>();
+ boolean _failReingestion = false;
FakePinotLLCRealtimeSegmentManager() {
this(createMockedResourceManager());
@@ -3054,7 +3306,12 @@ public class PinotLLCRealtimeSegmentManagerTest {
}
FakePinotLLCRealtimeSegmentManager(PinotHelixResourceManager
pinotHelixResourceManager) {
- super(pinotHelixResourceManager, CONTROLLER_CONF,
mock(ControllerMetrics.class));
+ this(pinotHelixResourceManager, mock(ControllerMetrics.class));
+ }
+
+ FakePinotLLCRealtimeSegmentManager(PinotHelixResourceManager
pinotHelixResourceManager,
+ ControllerMetrics controllerMetrics) {
+ super(pinotHelixResourceManager, CONTROLLER_CONF, controllerMetrics);
_mockResourceManager = pinotHelixResourceManager;
}
@@ -3285,6 +3542,15 @@ public class PinotLLCRealtimeSegmentManagerTest {
long getCurrentTimeMs() {
return _currentTimeMs;
}
+
+ @Override
+ void triggerReingestion(String serverHostPort, String segmentName)
+ throws IOException {
+ if (_failReingestion) {
+ throw new IOException("Injected re-ingestion failure for segment: " +
segmentName);
+ }
+ _reingestionRequests.put(segmentName, serverHostPort);
+ }
}
private static class FakePinotLLCRealtimeSegmentManagerII extends
FakePinotLLCRealtimeSegmentManager {
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PauselessDedupErrorReplicaRepairTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PauselessDedupErrorReplicaRepairTest.java
new file mode 100644
index 00000000000..578bb549c06
--- /dev/null
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PauselessDedupErrorReplicaRepairTest.java
@@ -0,0 +1,281 @@
+/**
+ * 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.integration.tests.realtime.utils.PauselessRealtimeTestUtils;
+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(alwaysRun = true)
+ 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, i.e. reaches its ideal state: leaving ERROR is not enough,
since a reset replica goes through OFFLINE.
+ // 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
(PauselessRealtimeTestUtils.isExternalViewConverged(_helixResourceManager,
realtimeTableName)) {
+ 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(alwaysRun = true)
+ public void tearDown()
+ throws Exception {
+ // Run every step even if an earlier one fails, so a failed test does not
leak the cluster into later suites
+ Throwable cleanupFailure = null;
+ try {
+ if (_helixResourceManager != null &&
_helixResourceManager.getRealtimeTableConfig(getTableName()) != null) {
+ dropRealtimeTable(getTableName());
+ }
+ } catch (Throwable t) {
+ cleanupFailure = addCleanupFailure(cleanupFailure, t);
+ }
+ try {
+ if (!_serverStarters.isEmpty()) {
+ stopServer();
+ }
+ } catch (Throwable t) {
+ cleanupFailure = addCleanupFailure(cleanupFailure, t);
+ }
+ try {
+ if (!_brokerStarters.isEmpty()) {
+ stopBroker();
+ }
+ } catch (Throwable t) {
+ cleanupFailure = addCleanupFailure(cleanupFailure, t);
+ }
+ try {
+ if (_controllerStarter != null) {
+ stopController();
+ }
+ } catch (Throwable t) {
+ cleanupFailure = addCleanupFailure(cleanupFailure, t);
+ }
+ try {
+ stopKafka();
+ } catch (Throwable t) {
+ cleanupFailure = addCleanupFailure(cleanupFailure, t);
+ }
+ try {
+ stopZk();
+ } catch (Throwable t) {
+ cleanupFailure = addCleanupFailure(cleanupFailure, t);
+ }
+ try {
+ FileUtils.deleteDirectory(_tempDir);
+ } catch (Throwable t) {
+ cleanupFailure = addCleanupFailure(cleanupFailure, t);
+ }
+ if (cleanupFailure != null) {
+ rethrowCleanupFailure(cleanupFailure);
+ }
+ }
+
+ private static Throwable addCleanupFailure(Throwable cleanupFailure,
Throwable failure) {
+ if (cleanupFailure == null) {
+ return failure;
+ }
+ cleanupFailure.addSuppressed(failure);
+ return cleanupFailure;
+ }
+
+ private static void rethrowCleanupFailure(Throwable cleanupFailure)
+ throws Exception {
+ if (cleanupFailure instanceof Error) {
+ throw (Error) cleanupFailure;
+ }
+ if (cleanupFailure instanceof Exception) {
+ throw (Exception) cleanupFailure;
+ }
+ throw new RuntimeException(cleanupFailure);
+ }
+}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/utils/PauselessRealtimeTestUtils.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/utils/PauselessRealtimeTestUtils.java
index 4b117234f01..783fbc2f156 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/utils/PauselessRealtimeTestUtils.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/realtime/utils/PauselessRealtimeTestUtils.java
@@ -55,19 +55,25 @@ public class PauselessRealtimeTestUtils {
/// `CONSUMING -> ONLINE` transition would make the one-shot repair
permanently miss the segment.
public static void waitForExternalViewToConverge(PinotHelixResourceManager
helixResourceManager,
String tableNameWithType, long timeoutMs) {
- TestUtils.waitForCondition(aVoid -> {
- IdealState idealState =
helixResourceManager.getTableIdealState(tableNameWithType);
- ExternalView externalView =
helixResourceManager.getTableExternalView(tableNameWithType);
- if (idealState == null || externalView == null) {
+ TestUtils.waitForCondition(aVoid ->
isExternalViewConverged(helixResourceManager, tableNameWithType), 1000,
+ timeoutMs, "External view failed to converge to ideal state for table:
" + tableNameWithType);
+ }
+
+ /// Returns whether every instance of every segment in the ideal state of
the given table is in its prescribed state
+ /// in the external view.
+ public static boolean isExternalViewConverged(PinotHelixResourceManager
helixResourceManager,
+ String tableNameWithType) {
+ IdealState idealState =
helixResourceManager.getTableIdealState(tableNameWithType);
+ ExternalView externalView =
helixResourceManager.getTableExternalView(tableNameWithType);
+ if (idealState == null || externalView == null) {
+ return false;
+ }
+ for (Map.Entry<String, Map<String, String>> entry :
idealState.getRecord().getMapFields().entrySet()) {
+ if (!entry.getValue().equals(externalView.getStateMap(entry.getKey()))) {
return false;
}
- for (Map.Entry<String, Map<String, String>> entry :
idealState.getRecord().getMapFields().entrySet()) {
- if
(!entry.getValue().equals(externalView.getStateMap(entry.getKey()))) {
- return false;
- }
- }
- return true;
- }, 1000, timeoutMs, "External view failed to converge to ideal state for
table: " + tableNameWithType);
+ }
+ return true;
}
/// Marks all current segments of the given table as exceeding the max
segment completion time, making them
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
