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 edc48dff392 Let a retried validation-manager IdealState update add the
segments it created (#19739)
edc48dff392 is described below
commit edc48dff392553e812c516efb3ba890580e96965
Author: Suvodeep Pyne <[email protected]>
AuthorDate: Wed Oct 21 01:25:34 2026 -0700
Let a retried validation-manager IdealState update add the segments it
created (#19739)
* Let a retried validation-manager IdealState update add the segments it
created
RealtimeSegmentValidationManager repairs partitions inside an IdealState
updater: it creates the ZK metadata of the next CONSUMING segment, then adds
the segment to the IdealState. When the IdealState write fails, e.g. on a
version conflict with another writer, the updater runs again. The new
segment's ZK metadata is still there, so the retry finds a latest segment
that
has ZK metadata but is not in the IdealState, and skips it because its ZK
metadata was updated less than max.segment.completion.time.millis ago (the
check that protects segments being committed). The repair lands only on a
later run after that time, while the partition does not consume.
Track the segments created by the updater across its attempts, and add a
segment created by an earlier attempt to the IdealState right away. A
segment
is tracked as soon as its ZK metadata is created, before instances are
assigned to it. The potential data loss is still reported when the segment
replaces a CONSUMING segment that is no longer CONSUMING on the retry, but
not for segments created for OFFLINE, ONLINE or new partitions, which have
no
previous CONSUMING segment.
* Check the retry's created segments through ZK metadata
testEnsureAllPartitionsConsumingRetryAddsSegmentsCreatedByEarlierAttempt
asserted the contents of the internal map that tracks the segments created
across attempts. Check the outcome instead: after the retry, partitions 0
and
1 have exactly the expected segment ZK metadata, so no extra segment was
created, and the IdealState assertions cover the rest.
* Prefer the IdealState segment among same-sequence repair candidates
When two controllers repair a partition at the same time, e.g. during a
lead controller change, both can create a new segment with the same sequence
number. If the other controller adds its segment to the IdealState first, the
retry of the IdealState update could pick its own new segment as the latest
one, add it right away as created by an earlier attempt, and fail on the
duplicate sequence number check, which aborts the whole update. The latest
segment of a partition is now the one [...]
Also covers the public ensureAllPartitionsConsuming with a failed
IdealState write, and the retry for a new partition.
---
.../realtime/PinotLLCRealtimeSegmentManager.java | 50 ++++--
.../PinotLLCRealtimeSegmentManagerTest.java | 196 ++++++++++++++++++++-
2 files changed, 233 insertions(+), 13 deletions(-)
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 b4add6ac69e..d3211b55feb 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
@@ -1548,8 +1548,9 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
/// Returns the latest LLC realtime segment ZK metadata for each partition.
///
/// When several segments of a partition have the highest sequence number,
e.g. the new segment of a failed commit
- /// start and the new segment of a concurrent repair, the one in the
IdealState is the latest one. If none of them is
- /// in the IdealState, whichever is listed first is the latest one.
+ /// start and the new segment of a concurrent repair, or the new segments of
two controllers repairing the partition
+ /// at the same time, the one in the IdealState is the latest one. If none
of them is in the IdealState, whichever is
+ /// listed first is the latest one.
///
/// @param realtimeTableName Realtime table name
/// @param instanceStatesMap Instance states map of the IdealState
@@ -1622,6 +1623,10 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
Preconditions.checkState(!_isStopping, "Segment manager is stopping");
String realtimeTableName = tableConfig.getTableName();
+ // The updater below runs again when the IdealState write fails (e.g. on a
version conflict), while the segment ZK
+ // metadata created by the failed attempt remains. Track the created
segments so that the next attempt still adds
+ // them to the IdealState.
+ Map<String, Boolean> newSegmentsCreated = new ConcurrentHashMap<>();
try {
HelixHelper.updateIdealState(_helixManager, realtimeTableName,
idealState -> {
assert idealState != null;
@@ -1643,7 +1648,7 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
getNewStreamMetadataList(streamConfigs,
currentPartitionGroupConsumptionStatusList, idealState);
streamConfigs.stream().forEach(streamConfig ->
streamConfig.setOffsetCriteria(originalOffsetCriteria));
return ensureAllPartitionsConsuming(tableConfig, streamConfigs,
idealState, streamMetadataList,
- offsetCriteria);
+ offsetCriteria, newSegmentsCreated);
} else {
LOGGER.info("Skipping LLC segments validation for table: {},
isTableEnabled: {}, isTablePaused: {}",
realtimeTableName, isTableEnabled, isTablePaused);
@@ -1914,6 +1919,16 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
@VisibleForTesting
IdealState ensureAllPartitionsConsuming(TableConfig tableConfig,
List<StreamConfig> streamConfigs,
IdealState idealState, List<StreamMetadata> streamMetadataList,
OffsetCriteria offsetCriteria) {
+ return ensureAllPartitionsConsuming(tableConfig, streamConfigs,
idealState, streamMetadataList, offsetCriteria,
+ new HashMap<>());
+ }
+
+ /// @param newSegmentsCreated Segments created by earlier attempts of the
same IdealState update, mapped to whether
+ /// they replace a CONSUMING segment. Segments
created by this attempt are added to it.
+ @VisibleForTesting
+ IdealState ensureAllPartitionsConsuming(TableConfig tableConfig,
List<StreamConfig> streamConfigs,
+ IdealState idealState, List<StreamMetadata> streamMetadataList,
OffsetCriteria offsetCriteria,
+ Map<String, Boolean> newSegmentsCreated) {
String realtimeTableName = tableConfig.getTableName();
InstancePartitions instancePartitions =
getConsumingInstancePartitions(tableConfig);
@@ -2006,6 +2021,7 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
createNewSegmentZKMetadata(tableConfig,
streamConfigs.get(streamConfigIdx), newLLCSegmentName,
currentTimeMs,
committingSegmentDescriptor, latestSegmentZKMetadata,
instancePartitions, numPartitions, numReplicas);
+ newSegmentsCreated.put(newSegmentName, true);
updateInstanceStatesForNewConsumingSegment(instanceStatesMap,
latestSegmentName, newSegmentName,
segmentAssignment, instancePartitionsMap);
} else { // partition group reached end of life
@@ -2078,7 +2094,7 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
latestSegmentZKMetadata.getStartOffset()); // segments are
OFFLINE; start from beginning
createNewConsumingSegment(tableConfig,
streamConfigs.get(streamConfigIdx), latestSegmentZKMetadata,
currentTimeMs, numPartitions, instancePartitions,
instanceStatesMap, segmentAssignment,
- instancePartitionsMap, startOffset);
+ instancePartitionsMap, startOffset, newSegmentsCreated);
} else {
LOGGER.info("Resuming consumption for partition: {} of table: {}",
partitionId, realtimeTableName);
StreamPartitionMsgOffset startOffset =
@@ -2086,7 +2102,7 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
tableConfig.getTableName(), offsetFactory,
latestSegmentZKMetadata.getEndOffset());
createNewConsumingSegment(tableConfig,
streamConfigs.get(streamConfigIdx), latestSegmentZKMetadata,
currentTimeMs, numPartitions, instancePartitions,
instanceStatesMap, segmentAssignment,
- instancePartitionsMap, startOffset);
+ instancePartitionsMap, startOffset, newSegmentsCreated);
}
}
} else {
@@ -2095,11 +2111,19 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
// i.e. after updating old segment metadata (old segment metadata
state = DONE/COMMITTING)
// and creating new segment metadata (new segment metadata state =
IN_PROGRESS),
// but before updating ideal state (new segment ideal missing from
ideal state)
- if (!isExceededMaxSegmentCompletionTime(realtimeTableName,
latestSegmentName, currentTimeMs)) {
- continue;
+ // The segment can also be created by an earlier attempt of this
update, which is not committing anything and
+ // can be added right away.
+ Boolean replacesConsumingSegment =
newSegmentsCreated.get(latestSegmentName);
+ boolean createdByEarlierAttempt = replacesConsumingSegment != null;
+ if (createdByEarlierAttempt) {
+ LOGGER.info("Adding segment: {} created by an earlier attempt of the
IdealState update", latestSegmentName);
+ } else {
+ if (!isExceededMaxSegmentCompletionTime(realtimeTableName,
latestSegmentName, currentTimeMs)) {
+ continue;
+ }
+ LOGGER.info("Repairing segment: {} which has segment ZK metadata but
does not exist in IdealState",
+ latestSegmentName);
}
- LOGGER.info("Repairing segment: {} which has segment ZK metadata but
does not exist in IdealState",
- latestSegmentName);
if (latestSegmentZKMetadata.getStatus() == Status.IN_PROGRESS) {
// Find the previous CONSUMING segment
@@ -2111,7 +2135,8 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
break;
}
}
- if (previousConsumingSegment == null) {
+ // A segment created for an OFFLINE, ONLINE or new partition has no
previous CONSUMING segment
+ if (previousConsumingSegment == null && (!createdByEarlierAttempt ||
replacesConsumingSegment)) {
LOGGER.error(
"Failed to find previous CONSUMING segment for partition: {}
of table: {}, potential data loss",
partitionId, realtimeTableName);
@@ -2134,6 +2159,7 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
String newSegmentName =
setupNewPartitionGroup(tableConfig,
streamMetadata.getStreamConfig(), partitionGroupMetadata,
currentTimeMs, instancePartitions, numPartitions,
numReplicas);
+ newSegmentsCreated.put(newSegmentName, false);
updateInstanceStatesForNewConsumingSegment(instanceStatesMap, null,
newSegmentName, segmentAssignment,
instancePartitionsMap);
}
@@ -2147,7 +2173,8 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
SegmentZKMetadata latestSegmentZKMetadata, long currentTimeMs,
int numPartitions, InstancePartitions instancePartitions,
Map<String, Map<String, String>> instanceStatesMap, SegmentAssignment
segmentAssignment,
- Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap,
StreamPartitionMsgOffset startOffset) {
+ Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap,
StreamPartitionMsgOffset startOffset,
+ Map<String, Boolean> newSegmentsCreated) {
int numReplicas = getNumReplicas(tableConfig, instancePartitions);
LLCSegmentName latestLLCSegmentName = new
LLCSegmentName(latestSegmentZKMetadata.getSegmentName());
LLCSegmentName newLLCSegmentName =
getNextLLCSegmentName(latestLLCSegmentName, currentTimeMs);
@@ -2156,6 +2183,7 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
createNewSegmentZKMetadata(tableConfig, streamConfig, newLLCSegmentName,
currentTimeMs, committingSegmentDescriptor,
latestSegmentZKMetadata, instancePartitions, numPartitions,
numReplicas);
String newSegmentName = newLLCSegmentName.getSegmentName();
+ newSegmentsCreated.put(newSegmentName, false);
updateInstanceStatesForNewConsumingSegment(instanceStatesMap, null,
newSegmentName, segmentAssignment,
instancePartitionsMap);
}
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 2f943077ccb..4eeef13372e 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
@@ -27,6 +27,7 @@ import java.net.URISyntaxException;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
@@ -41,6 +42,7 @@ import java.util.TreeMap;
import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import javax.annotation.Nullable;
@@ -63,6 +65,7 @@ import
org.apache.pinot.common.exception.HttpErrorStatusException;
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.ControllerMeter;
import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.common.restlet.resources.BatchConfig;
import org.apache.pinot.common.restlet.resources.PauseStatusDetails;
@@ -70,6 +73,7 @@ import
org.apache.pinot.common.restlet.resources.TableLLCSegmentUploadResponse;
import org.apache.pinot.common.utils.FileUploadDownloadClient;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.common.utils.URIUtils;
+import org.apache.pinot.common.utils.helix.HelixHelper;
import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import
org.apache.pinot.controller.helix.core.assignment.segment.SegmentAssignment;
@@ -117,6 +121,7 @@ import org.joda.time.Interval;
import org.mockito.MockedStatic;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import static
org.apache.pinot.controller.ControllerConf.ControllerPeriodicTasksConf.ENABLE_TMP_SEGMENT_ASYNC_DELETION;
@@ -1029,6 +1034,181 @@ public class PinotLLCRealtimeSegmentManagerTest {
assertEquals(consumingSegmentInstanceStateMap.get(offlineInstance),
SegmentStateModel.CONSUMING);
}
+ @Test
+ public void
testEnsureAllPartitionsConsumingRetryAddsSegmentsCreatedByEarlierAttempt() {
+ ControllerMetrics controllerMetrics = mock(ControllerMetrics.class);
+ FakePinotLLCRealtimeSegmentManager segmentManager = new
FakePinotLLCRealtimeSegmentManager(controllerMetrics);
+ setUpNewTable(segmentManager, 2, 5, 4);
+ Map<String, Map<String, String>> instanceStatesMap =
segmentManager._idealState.getRecord().getMapFields();
+
+ // Partition 0: step 1 of the segment commit is done, but the segment is
still CONSUMING in IdealState
+ String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0,
CURRENT_TIME_MS).getSegmentName();
+ SegmentZKMetadata committingSegmentZKMetadata =
segmentManager._segmentZKMetadataMap.get(committingSegment);
+ committingSegmentZKMetadata.setStatus(Status.DONE);
+ committingSegmentZKMetadata.setEndOffset(NEXT_OFFSET);
+
segmentManager._segmentsExceedingMaxSegmentCompletionTime.add(committingSegment);
+ // Partition 1: all replicas of the consuming segment are OFFLINE
+ String offlineSegment = new LLCSegmentName(RAW_TABLE_NAME, 1, 0,
CURRENT_TIME_MS).getSegmentName();
+ instanceStatesMap.get(offlineSegment).replaceAll((instance, state) ->
SegmentStateModel.OFFLINE);
+ // Partition 4: a new partition of the stream
+ segmentManager._numPartitions = 5;
+
+ // The first attempt creates the new segments, but its IdealState write
fails
+ Map<String, Boolean> newSegmentsCreated = new HashMap<>();
+ runFailedEnsureAllPartitionsConsumingAttempt(segmentManager,
newSegmentsCreated);
+ String newSegment0 = new LLCSegmentName(RAW_TABLE_NAME, 0, 1,
CURRENT_TIME_MS).getSegmentName();
+ String newSegment1 = new LLCSegmentName(RAW_TABLE_NAME, 1, 1,
CURRENT_TIME_MS).getSegmentName();
+ String newSegment4 = new LLCSegmentName(RAW_TABLE_NAME, 4, 0,
CURRENT_TIME_MS).getSegmentName();
+
+ // The retry starts from the unchanged IdealState. It must add the new
segments right away instead of waiting for
+ // them to exceed the max segment completion time.
+ segmentManager.ensureAllPartitionsConsuming(segmentManager._tableConfig,
segmentManager._streamConfigs,
+ segmentManager._idealState, getStreamMetadataList(segmentManager),
null, newSegmentsCreated);
+ assertEquals(segmentManager._segmentZKMetadataMap.keySet().stream()
+ .filter(segment -> Set.of(0, 1, 4).contains(new
LLCSegmentName(segment).getPartitionGroupId()))
+ .collect(Collectors.toSet()),
+ Set.of(committingSegment, newSegment0, offlineSegment, newSegment1,
newSegment4));
+ assertEquals(new
HashSet<>(instanceStatesMap.get(committingSegment).values()),
Set.of(SegmentStateModel.ONLINE));
+ assertEquals(new HashSet<>(instanceStatesMap.get(newSegment0).values()),
Set.of(SegmentStateModel.CONSUMING));
+ assertEquals(new
HashSet<>(instanceStatesMap.get(offlineSegment).values()),
Set.of(SegmentStateModel.OFFLINE));
+ assertEquals(new HashSet<>(instanceStatesMap.get(newSegment1).values()),
Set.of(SegmentStateModel.CONSUMING));
+ assertEquals(new HashSet<>(instanceStatesMap.get(newSegment4).values()),
Set.of(SegmentStateModel.CONSUMING));
+ verify(controllerMetrics,
never()).addMeteredTableValue(REALTIME_TABLE_NAME,
ControllerMeter.LLC_STREAM_DATA_LOSS,
+ 1L);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void
testEnsureAllPartitionsConsumingAddsSegmentsCreatedByAttemptWithFailedIdealStateWrite()
{
+ PinotHelixResourceManager resourceManager =
FakePinotLLCRealtimeSegmentManager.createMockedResourceManager();
+ FakePinotLLCRealtimeSegmentManager segmentManager = new
FakePinotLLCRealtimeSegmentManager(resourceManager);
+ setUpNewTable(segmentManager, 2, 5, 4);
+ String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0,
CURRENT_TIME_MS).getSegmentName();
+ SegmentZKMetadata committingSegmentZKMetadata =
segmentManager._segmentZKMetadataMap.get(committingSegment);
+ committingSegmentZKMetadata.setStatus(Status.DONE);
+ committingSegmentZKMetadata.setEndOffset(NEXT_OFFSET);
+
segmentManager._segmentsExceedingMaxSegmentCompletionTime.add(committingSegment);
+
+ // The first IdealState write fails because another controller uploads a
segment in the meantime
+ HelixDataAccessor dataAccessor = mock(HelixDataAccessor.class);
+
when(resourceManager.getHelixZkManager().getHelixDataAccessor()).thenReturn(dataAccessor);
+ when(dataAccessor.keyBuilder()).thenReturn(new
PropertyKey.Builder(CLUSTER_NAME));
+ when(dataAccessor.getProperty(any(PropertyKey.class))).thenAnswer(
+ invocation -> HelixHelper.cloneIdealState(segmentManager._idealState));
+ BaseDataAccessor<ZNRecord> baseDataAccessor = mock(BaseDataAccessor.class);
+ when(dataAccessor.getBaseDataAccessor()).thenReturn(baseDataAccessor);
+ String uploadedSegment = "uploadedSegment";
+ AtomicInteger numWrites = new AtomicInteger();
+ when(baseDataAccessor.set(anyString(), any(ZNRecord.class), anyInt(),
anyInt())).thenAnswer(invocation -> {
+ if (numWrites.getAndIncrement() == 0) {
+ segmentManager._idealState.getRecord().getMapFields()
+ .put(uploadedSegment, Map.of(Helix.PREFIX_OF_SERVER_INSTANCE + 0,
SegmentStateModel.ONLINE));
+ return false;
+ }
+ segmentManager._idealState = new
IdealState(invocation.<ZNRecord>getArgument(1));
+ return true;
+ });
+
+ // The retry adds the new segment created by the first attempt right away,
and keeps the uploaded segment
+ segmentManager.ensureAllPartitionsConsuming(segmentManager._tableConfig,
segmentManager._streamConfigs, null);
+ assertEquals(numWrites.get(), 2);
+ String newSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 1,
CURRENT_TIME_MS).getSegmentName();
+ assertEquals(segmentManager._segmentZKMetadataMap.keySet().stream()
+ .filter(segment -> new LLCSegmentName(segment).getPartitionGroupId()
== 0)
+ .collect(Collectors.toSet()), Set.of(committingSegment, newSegment));
+ Map<String, Map<String, String>> instanceStatesMap =
segmentManager._idealState.getRecord().getMapFields();
+ assertEquals(new
HashSet<>(instanceStatesMap.get(committingSegment).values()),
Set.of(SegmentStateModel.ONLINE));
+ assertEquals(new HashSet<>(instanceStatesMap.get(newSegment).values()),
Set.of(SegmentStateModel.CONSUMING));
+ assertTrue(instanceStatesMap.containsKey(uploadedSegment));
+ }
+
+ @DataProvider
+ public Object[][] reverseLLCSegments() {
+ return new Object[][]{{false}, {true}};
+ }
+
+ @Test(dataProvider = "reverseLLCSegments")
+ public void
testEnsureAllPartitionsConsumingRetryKeepsSegmentAddedByAnotherController(boolean
reverseLLCSegments) {
+ ControllerMetrics controllerMetrics = mock(ControllerMetrics.class);
+ FakePinotLLCRealtimeSegmentManager segmentManager = new
FakePinotLLCRealtimeSegmentManager(controllerMetrics);
+ setUpNewTable(segmentManager, 2, 5, 4);
+ // Lists the segments in both orders across the test runs, as the segments
with the same sequence number below
+ // must be resolved the same way in either order
+ segmentManager._reverseLLCSegments = reverseLLCSegments;
+ Map<String, Map<String, String>> instanceStatesMap =
segmentManager._idealState.getRecord().getMapFields();
+ String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0,
CURRENT_TIME_MS).getSegmentName();
+ SegmentZKMetadata committingSegmentZKMetadata =
segmentManager._segmentZKMetadataMap.get(committingSegment);
+ committingSegmentZKMetadata.setStatus(Status.DONE);
+ committingSegmentZKMetadata.setEndOffset(NEXT_OFFSET);
+
segmentManager._segmentsExceedingMaxSegmentCompletionTime.add(committingSegment);
+
+ // The first attempt creates a new segment, but its IdealState write fails
+ Map<String, Boolean> newSegmentsCreated = new HashMap<>();
+ runFailedEnsureAllPartitionsConsumingAttempt(segmentManager,
newSegmentsCreated);
+ String newSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 1,
CURRENT_TIME_MS).getSegmentName();
+
+ // Meanwhile, another controller, e.g. the previous lead controller,
creates its own new segment with the same
+ // sequence number and adds it to the IdealState first
+ String otherNewSegment =
+ new LLCSegmentName(RAW_TABLE_NAME, 0, 1, CURRENT_TIME_MS -
TimeUnit.MINUTES.toMillis(1)).getSegmentName();
+ segmentManager.persistSegmentZKMetadata(REALTIME_TABLE_NAME, new
SegmentZKMetadata(
+ new
ZNRecord(segmentManager._segmentZKMetadataMap.get(newSegment).toZNRecord(),
otherNewSegment)), -1);
+ Map<String, String> otherNewSegmentInstanceStateMap = new TreeMap<>();
+ for (String instance : instanceStatesMap.get(committingSegment).keySet()) {
+ otherNewSegmentInstanceStateMap.put(instance,
SegmentStateModel.CONSUMING);
+ }
+ instanceStatesMap.get(committingSegment).replaceAll((instance, state) ->
SegmentStateModel.ONLINE);
+ instanceStatesMap.put(otherNewSegment, otherNewSegmentInstanceStateMap);
+
+ // The retry keeps the segment of the other controller instead of adding
its own one
+ Map<String, Map<String, String>> expectedInstanceStatesMap =
cloneInstanceStatesMap(instanceStatesMap);
+ Set<String> expectedSegments = new
HashSet<>(segmentManager._segmentZKMetadataMap.keySet());
+ segmentManager.ensureAllPartitionsConsuming(segmentManager._tableConfig,
segmentManager._streamConfigs,
+ segmentManager._idealState, getStreamMetadataList(segmentManager),
null, newSegmentsCreated);
+ assertEquals(instanceStatesMap, expectedInstanceStatesMap);
+ assertEquals(segmentManager._segmentZKMetadataMap.keySet(),
expectedSegments);
+ verify(controllerMetrics,
never()).addMeteredTableValue(REALTIME_TABLE_NAME,
ControllerMeter.LLC_STREAM_DATA_LOSS,
+ 1L);
+ }
+
+ @Test
+ public void
testEnsureAllPartitionsConsumingRetryReportsDataLossWhenReplacedSegmentGoesOffline()
{
+ ControllerMetrics controllerMetrics = mock(ControllerMetrics.class);
+ FakePinotLLCRealtimeSegmentManager segmentManager = new
FakePinotLLCRealtimeSegmentManager(controllerMetrics);
+ setUpNewTable(segmentManager, 2, 5, 4);
+ Map<String, Map<String, String>> instanceStatesMap =
segmentManager._idealState.getRecord().getMapFields();
+ String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0,
CURRENT_TIME_MS).getSegmentName();
+ SegmentZKMetadata committingSegmentZKMetadata =
segmentManager._segmentZKMetadataMap.get(committingSegment);
+ committingSegmentZKMetadata.setStatus(Status.DONE);
+ committingSegmentZKMetadata.setEndOffset(NEXT_OFFSET);
+
segmentManager._segmentsExceedingMaxSegmentCompletionTime.add(committingSegment);
+
+ Map<String, Boolean> newSegmentsCreated = new HashMap<>();
+ runFailedEnsureAllPartitionsConsumingAttempt(segmentManager,
newSegmentsCreated);
+
+ // All replicas of the segment being replaced go OFFLINE before the retry
+ instanceStatesMap.get(committingSegment).replaceAll((instance, state) ->
SegmentStateModel.OFFLINE);
+ segmentManager.ensureAllPartitionsConsuming(segmentManager._tableConfig,
segmentManager._streamConfigs,
+ segmentManager._idealState, getStreamMetadataList(segmentManager),
null, newSegmentsCreated);
+ String newSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 1,
CURRENT_TIME_MS).getSegmentName();
+ assertEquals(new HashSet<>(instanceStatesMap.get(newSegment).values()),
Set.of(SegmentStateModel.CONSUMING));
+ verify(controllerMetrics).addMeteredTableValue(REALTIME_TABLE_NAME,
ControllerMeter.LLC_STREAM_DATA_LOSS, 1L);
+ }
+
+ /// Runs ensureAllPartitionsConsuming on a copy of the IdealState, as an
attempt whose IdealState write fails.
+ private void
runFailedEnsureAllPartitionsConsumingAttempt(FakePinotLLCRealtimeSegmentManager
segmentManager,
+ Map<String, Boolean> newSegmentsCreated) {
+ IdealState failedAttemptIdealState = new IdealState(REALTIME_TABLE_NAME);
+ failedAttemptIdealState.getRecord()
+
.setMapFields(cloneInstanceStatesMap(segmentManager._idealState.getRecord().getMapFields()));
+ segmentManager.ensureAllPartitionsConsuming(segmentManager._tableConfig,
segmentManager._streamConfigs,
+ failedAttemptIdealState, getStreamMetadataList(segmentManager), null,
newSegmentsCreated);
+ }
+
+ private List<StreamMetadata>
getStreamMetadataList(FakePinotLLCRealtimeSegmentManager segmentManager) {
+ return
segmentManager.getNewStreamMetadataList(segmentManager._streamConfigs,
List.of(), mock(IdealState.class));
+ }
+
@Test
public void testPartialOfflineReplicaRepairDisabled() {
// Set up a new table with 3 replicas, 5 instances, 2 partitions
@@ -2843,6 +3023,10 @@ public class PinotLLCRealtimeSegmentManagerTest {
int _numPartitions;
List<StreamMetadata> _streamMetadataList = null;
boolean _exceededMaxSegmentCompletionTime = false;
+ // Segments that exceed the max segment completion time even when
_exceededMaxSegmentCompletionTime is false
+ Set<String> _segmentsExceedingMaxSegmentCompletionTime = new HashSet<>();
+ // Whether to list the LLC segments in reverse order
+ boolean _reverseLLCSegments = false;
// Number of upcoming IdealState updates on segment completion that fail
without being applied
int _numFailedIdealStateUpdates = 0;
// Number of upcoming IdealState updates on segment completion that are
applied but reported as failed
@@ -2874,6 +3058,10 @@ public class PinotLLCRealtimeSegmentManagerTest {
_mockResourceManager = pinotHelixResourceManager;
}
+ FakePinotLLCRealtimeSegmentManager(ControllerMetrics controllerMetrics) {
+ super(createMockedResourceManager(), CONTROLLER_CONF, controllerMetrics);
+ }
+
FakePinotLLCRealtimeSegmentManager(PinotHelixResourceManager
pinotHelixResourceManager,
boolean enablePartialOfflineReplicaRepair) {
super(pinotHelixResourceManager,
createControllerConf(enablePartialOfflineReplicaRepair),
@@ -2964,7 +3152,11 @@ public class PinotLLCRealtimeSegmentManagerTest {
@Override
List<String> getLLCSegments(String realtimeTableName) {
- return new ArrayList<>(_segmentZKMetadataMap.keySet());
+ List<String> llcSegments = new
ArrayList<>(_segmentZKMetadataMap.keySet());
+ if (_reverseLLCSegments) {
+ Collections.reverse(llcSegments);
+ }
+ return llcSegments;
}
@Override
@@ -3086,7 +3278,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
@Override
protected boolean isExceededMaxSegmentCompletionTime(String
realtimeTableName, String segmentName,
long currentTimeMs) {
- return _exceededMaxSegmentCompletionTime;
+ return _exceededMaxSegmentCompletionTime ||
_segmentsExceedingMaxSegmentCompletionTime.contains(segmentName);
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
