This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang 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 8cf16156a44 Remove the kept new segment when a resumed pauseless
commit start completes without it (#19995)
8cf16156a44 is described below
commit 8cf16156a44eb87096f648a9832bb091d3509710
Author: Suvodeep Pyne <[email protected]>
AuthorDate: Fri Oct 23 15:34:46 2026 -0700
Remove the kept new segment when a resumed pauseless commit start completes
without it (#19995)
---
.../helix/core/PinotHelixResourceManager.java | 89 +++++++++++
.../realtime/PinotLLCRealtimeSegmentManager.java | 82 ++++++----
.../PinotHelixResourceManagerStatelessTest.java | 135 +++++++++++++++++
.../PinotLLCRealtimeSegmentManagerTest.java | 166 +++++++++++++++++++++
4 files changed, 447 insertions(+), 25 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 67793bec133..bd2dd0388bf 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
@@ -91,6 +91,8 @@ import org.apache.helix.store.zk.ZkHelixPropertyStore;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
import org.apache.helix.zookeeper.impl.client.ZkClient;
+import org.apache.helix.zookeeper.zkclient.exception.ZkException;
+import org.apache.helix.zookeeper.zkclient.exception.ZkNoNodeException;
import org.apache.pinot.common.assignment.InstanceAssignmentConfigUtils;
import org.apache.pinot.common.assignment.InstancePartitions;
import org.apache.pinot.common.assignment.InstancePartitionsUtils;
@@ -211,6 +213,8 @@ import org.apache.pinot.spi.utils.TimeUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.spi.utils.retry.RetryPolicies;
import org.apache.pinot.spi.utils.retry.RetryPolicy;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Op;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -221,6 +225,7 @@ public class PinotHelixResourceManager {
private static final long CACHE_ENTRY_EXPIRE_TIME_HOURS = 6L;
private static final RetryPolicy DEFAULT_RETRY_POLICY =
RetryPolicies.exponentialBackoffRetryPolicy(5, 1000L, 2.0f);
private static final int DEFAULT_SEGMENT_LINEAGE_UPDATE_NUM_RETRY = 5;
+ private static final int MAX_SEGMENT_ZK_METADATA_REMOVAL_ATTEMPTS = 5;
public static final String APPEND = "APPEND";
private static final int DEFAULT_LINEAGE_UPDATER_LOCKERS_SIZE = 500;
private static final String API_REQUEST_ID_PREFIX = "api-";
@@ -2278,6 +2283,90 @@ public class PinotHelixResourceManager {
}
}
+ /// Removes the ZK metadata of the given segments that are IN_PROGRESS and
not in the IdealState of the table, i.e.
+ /// new consuming segments that were not added to the IdealState, and
returns the segments whose ZK metadata was
+ /// removed.
+ ///
+ /// The removal is submitted in one ZK `multi()` transaction with
version-checked deletes of the segment ZK metadata
+ /// read here, and a version-checked write of the unchanged IdealState:
+ /// - An IdealState update that read the IdealState before the transaction
fails its version check. An updater that
+ /// reads the segment ZK metadata again when it retries, like the one of
RealtimeSegmentValidationManager, then no
+ /// longer finds the removed segments.
+ /// - If the IdealState, or the ZK metadata of one of the segments, is
updated after it was read here, the transaction
+ /// fails and is retried.
+ ///
+ /// The data versions cannot tell a segment ZK metadata deleted and
recreated with the same name since it was read,
+ /// e.g. by an upload under the same name, from the one that was read. As
the IdealState is outside the property
+ /// store, the transaction uses the multi-write ZkClient directly instead of
[ZkMultiWriteBuilder].
+ public Set<String> removeInProgressSegmentsZKMetadataNotInIdealState(String
tableNameWithType,
+ Collection<String> segmentNames) {
+ return
removeInProgressSegmentsZKMetadataNotInIdealState(tableNameWithType,
segmentNames, () -> {
+ });
+ }
+
+ /// @param afterRead Runs after reading the IdealState and the segment ZK
metadata, before removing it
+ @VisibleForTesting
+ Set<String> removeInProgressSegmentsZKMetadataNotInIdealState(String
tableNameWithType,
+ Collection<String> segmentNames, Runnable afterRead) {
+ ZkClient zkClient = getOrBuildMultiWriteZkClient();
+ String idealStatePath = PropertyPathBuilder.idealState(_helixClusterName,
tableNameWithType);
+ String propertyStoreRoot =
PropertyPathBuilder.propertyStore(_helixClusterName);
+ for (int attempt = 1; ; attempt++) {
+ Stat idealStateStat = new Stat();
+ ZNRecord idealState;
+ try {
+ idealState = zkClient.readData(idealStatePath, idealStateStat);
+ } catch (ZkNoNodeException e) {
+ idealState = null;
+ }
+ if (idealState == null) {
+ // The table was deleted
+ return Set.of();
+ }
+ List<Op> segmentDeletes = new ArrayList<>();
+ Set<String> segmentsToRemove = new HashSet<>();
+ for (String segmentName : segmentNames) {
+ if (idealState.getMapFields().containsKey(segmentName)) {
+ continue;
+ }
+ String segmentPath =
+ propertyStoreRoot +
ZKMetadataProvider.constructPropertyStorePathForSegment(tableNameWithType,
segmentName);
+ Stat segmentStat = new Stat();
+ ZNRecord segmentRecord;
+ try {
+ segmentRecord = zkClient.readData(segmentPath, segmentStat);
+ } catch (ZkNoNodeException e) {
+ continue;
+ }
+ if (new SegmentZKMetadata(segmentRecord).getStatus() ==
CommonConstants.Segment.Realtime.Status.IN_PROGRESS) {
+ segmentDeletes.add(Op.delete(segmentPath, segmentStat.getVersion()));
+ segmentsToRemove.add(segmentName);
+ }
+ }
+ if (segmentsToRemove.isEmpty()) {
+ return segmentsToRemove;
+ }
+ afterRead.run();
+ List<Op> ops = new ArrayList<>(segmentDeletes.size() + 1);
+ ops.add(Op.setData(idealStatePath, zkClient.serialize(idealState,
idealStatePath), idealStateStat.getVersion()));
+ ops.addAll(segmentDeletes);
+ try {
+ zkClient.multi(ops);
+ return segmentsToRemove;
+ } catch (ZkException e) {
+ // As in ZkMultiWriteBuilder#execute(), the ZkClient wraps the
KeeperException of an atomic rollback. A version
+ // mismatch or a missing node means that the IdealState or a segment
ZK metadata changed concurrently.
+ boolean concurrentChange = e.getCause() instanceof
KeeperException.BadVersionException
+ || e.getCause() instanceof KeeperException.NoNodeException;
+ if (!concurrentChange || attempt ==
MAX_SEGMENT_ZK_METADATA_REMOVAL_ATTEMPTS) {
+ throw e;
+ }
+ LOGGER.info("Retrying to remove segment ZK metadata for segments: {}
of table: {} after a concurrent change",
+ segmentsToRemove, tableNameWithType);
+ }
+ }
+ }
+
public boolean createZKNode(String path, ZNRecord record, int accessOption,
long ttl) {
return _helixDataAccessor.getBaseDataAccessor().create(path, record,
accessOption, ttl);
}
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 56d8ce41dd3..d2339fd5ad1 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
@@ -674,9 +674,16 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
// Step-2: Create new segment metadata if needed
long startTimeNs2 = System.nanoTime();
+ // Other attempts of the same commit start might have created the new
segment already, even several ones when
+ // attempts on 2 controllers overlap. Find them before deciding whether
the partition still needs a new segment, so
+ // that the unused ones can be removed.
+ List<String> newSegmentsOfEarlierAttempts =
+ resumed ? findNewSegmentsOfEarlierAttempts(realtimeTableName, new
LLCSegmentName(committingSegmentName))
+ : List.of();
String newConsumingSegmentName =
createNewSegmentMetadata(tableConfig, committingSegmentDescriptor,
committingSegmentZKMetadata,
- instancePartitions, toCommitting, resumed);
+ instancePartitions, toCommitting,
+ newSegmentsOfEarlierAttempts.isEmpty() ? null :
newSegmentsOfEarlierAttempts.get(0));
preProcessCommitIdealStateUpdate();
@@ -708,17 +715,20 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
}
Map<String, Map<String, String>> instanceStatesMapAfterStep3 =
idealState.getRecord().getMapFields();
- boolean newConsumingSegmentInIdealState = false;
+ boolean newConsumingSegmentInIdealState =
+ newConsumingSegmentName != null &&
instanceStatesMapAfterStep3.containsKey(newConsumingSegmentName);
+ // The committing segment is ONLINE now, so no attempt of the commit can
add a new segment that is not in the
+ // IdealState anymore. Remove their ZK metadata so that
RealtimeSegmentValidationManager does not add them either.
+ Set<String> unusedNewSegments = new
HashSet<>(newSegmentsOfEarlierAttempts);
if (newConsumingSegmentName != null) {
- newConsumingSegmentInIdealState =
instanceStatesMapAfterStep3.containsKey(newConsumingSegmentName);
- if (!newConsumingSegmentInIdealState) {
- LOGGER.info(
- "Cleaning up segment ZK metadata for new consuming segment {} of
table {} because it was not added to "
- + "IdealState. This can happen when table/topic consumption is
paused.",
- newConsumingSegmentName, realtimeTableName);
- removeSegmentZKMetadataBestEffort(realtimeTableName,
newConsumingSegmentName);
- newConsumingSegmentName = null;
- }
+ unusedNewSegments.add(newConsumingSegmentName);
+ }
+ unusedNewSegments.removeIf(instanceStatesMapAfterStep3::containsKey);
+ if (!unusedNewSegments.isEmpty()) {
+ removeUnusedNewSegmentsZKMetadata(realtimeTableName, unusedNewSegments);
+ }
+ if (!newConsumingSegmentInIdealState) {
+ newConsumingSegmentName = null;
}
long endTimeNs = System.nanoTime();
@@ -849,7 +859,7 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
return true;
}
- /// Returns the new segment created by an earlier attempt of a resumed
commit start, or `null` if there is none.
+ /// Returns the new segments created by earlier attempts of a resumed commit
start, oldest first.
///
/// The new segment is named after its creation time with minute
granularity, and is created between step 1 of the
/// first attempt (the modification time of the committing segment ZK
metadata) and now, which is bounded by the max
@@ -857,29 +867,34 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
/// clock differences between controllers and ZK. This relies on nothing
rewriting the committing segment ZK metadata
/// while it stays COMMITTING, as a later modification time would skip the
minutes in which the new segment was
/// created.
- @Nullable
- private String findNewSegmentOfEarlierAttempt(String realtimeTableName,
LLCSegmentName committingLLCSegment) {
+ private List<String> findNewSegmentsOfEarlierAttempts(String
realtimeTableName,
+ LLCSegmentName committingLLCSegment) {
Stat stat = new Stat();
getSegmentZKMetadata(realtimeTableName,
committingLLCSegment.getSegmentName(), stat);
long oneMinuteMs = TimeUnit.MINUTES.toMillis(1);
long startTimeMs = Math.floorDiv(stat.getMtime(), oneMinuteMs) *
oneMinuteMs - oneMinuteMs;
long endTimeMs = Math.floorDiv(getCurrentTimeMs(), oneMinuteMs) *
oneMinuteMs + oneMinuteMs;
+ List<String> newSegments = new ArrayList<>();
for (long timeMs = startTimeMs; timeMs <= endTimeMs; timeMs +=
oneMinuteMs) {
String segmentName = new
LLCSegmentName(committingLLCSegment.getTableName(),
committingLLCSegment.getPartitionGroupId(),
committingLLCSegment.getSequenceNumber() + 1,
timeMs).getSegmentName();
if (segmentZKMetadataExists(realtimeTableName, segmentName)) {
- return segmentName;
+ newSegments.add(segmentName);
}
}
- return null;
+ return newSegments;
}
// Step 2: Create new segment metadata
+
+ /// Returns the new consuming segment of the partition of the committing
segment, or `null` if the partition is no
+ /// longer in the stream, or the table/topic is paused. Uses
`newSegmentOfEarlierAttempt` if not `null`, and creates
+ /// the new segment ZK metadata otherwise.
@Nullable
private String createNewSegmentMetadata(TableConfig tableConfig,
CommittingSegmentDescriptor committingSegmentDescriptor,
SegmentZKMetadata committingSegmentZKMetadata,
- InstancePartitions instancePartitions, boolean createIfAbsent, boolean
resumed) {
+ InstancePartitions instancePartitions, boolean createIfAbsent, @Nullable
String newSegmentOfEarlierAttempt) {
String committingSegmentName =
committingSegmentDescriptor.getSegmentName();
String realtimeTableName = tableConfig.getTableName();
@@ -905,14 +920,10 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
+ "paused.", committingSegmentName);
return null;
}
- // Another attempt of the same commit start might have created the new
segment already, and the segment might be
- // consuming or even committed, so never overwrite its ZK metadata
- if (resumed) {
- String existingNewSegmentName =
findNewSegmentOfEarlierAttempt(realtimeTableName, committingLLCSegment);
- if (existingNewSegmentName != null) {
- LOGGER.info("Using existing segment metadata for new segment: {}",
existingNewSegmentName);
- return existingNewSegmentName;
- }
+ // The new segment of an earlier attempt might be consuming or even
committed, so never overwrite its ZK metadata
+ if (newSegmentOfEarlierAttempt != null) {
+ LOGGER.info("Using existing segment metadata for new segment: {}",
newSegmentOfEarlierAttempt);
+ return newSegmentOfEarlierAttempt;
}
String rawTableName =
TableNameBuilder.extractRawTableName(realtimeTableName);
long newSegmentCreationTimeMs = getCurrentTimeMs();
@@ -944,6 +955,27 @@ public class PinotLLCRealtimeSegmentManager implements
PinotClusterConfigChangeL
return newConsumingSegmentName;
}
+ /// Removes the ZK metadata of new segments that are not added to the
IdealState, e.g. when table/topic consumption is
+ /// paused, or the partition is no longer in the stream. The removal is
fenced against concurrent IdealState updates
+ /// that add one of them, e.g. by RealtimeSegmentValidationManager (see
+ ///
PinotHelixResourceManager#removeInProgressSegmentsZKMetadataNotInIdealState).
+ @VisibleForTesting
+ void removeUnusedNewSegmentsZKMetadata(String realtimeTableName, Set<String>
segmentNames) {
+ LOGGER.info("Cleaning up segment ZK metadata for new segments: {} of
table: {} which are not in the IdealState",
+ segmentNames, realtimeTableName);
+ try {
+ Set<String> removedSegments =
+
_helixResourceManager.removeInProgressSegmentsZKMetadataNotInIdealState(realtimeTableName,
segmentNames);
+ if (removedSegments.size() < segmentNames.size()) {
+ LOGGER.info("Removed segment ZK metadata for segments: {} of table:
{}, the others were added to the "
+ + "IdealState or removed concurrently", removedSegments,
realtimeTableName);
+ }
+ } catch (Exception e) {
+ LOGGER.warn("Caught exception while removing segment ZK metadata for
segments: {} of table: {}", segmentNames,
+ realtimeTableName, e);
+ }
+ }
+
private void removeSegmentZKMetadataBestEffort(String realtimeTableName,
String segmentName) {
String segmentMetadataPath =
ZKMetadataProvider.constructPropertyStorePathForSegment(realtimeTableName,
segmentName);
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerStatelessTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerStatelessTest.java
index d2d2dbc757f..7601a7779c4 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerStatelessTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerStatelessTest.java
@@ -31,7 +31,10 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.helix.AccessOption;
import org.apache.helix.HelixAdmin;
+import org.apache.helix.PropertyPathBuilder;
import
org.apache.helix.controller.dataproviders.ResourceControllerDataProvider;
import
org.apache.helix.controller.rebalancer.strategy.CrushEdRebalanceStrategy;
import org.apache.helix.model.ClusterConfig;
@@ -40,6 +43,7 @@ import org.apache.helix.model.IdealState;
import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.MasterSlaveSMD;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.helix.zookeeper.zkclient.exception.ZkBadVersionException;
import org.apache.pinot.common.exception.InvalidConfigException;
import org.apache.pinot.common.exception.TableNotFoundException;
import org.apache.pinot.common.lineage.LineageEntryState;
@@ -1896,6 +1900,137 @@ public class PinotHelixResourceManagerStatelessTest
extends ControllerTest {
assertTrue(ZKMetadataProvider.removeSegmentZKMetadata(_propertyStore,
OFFLINE_TABLE_NAME, segName2));
}
+ private static final String REMOVAL_TEST_TABLE =
"removeSegmentsZKMetadataTest_REALTIME";
+
+ /// Creates the IdealState of the removal test table with the given segment,
and IN_PROGRESS segment ZK metadata for
+ /// the given segments.
+ private void setUpRemovalTestTable(String segmentInIdealState, List<String>
segmentsWithZKMetadata) {
+ IdealState idealState = new IdealState(REMOVAL_TEST_TABLE);
+ idealState.setStateModelDefRef(
+
PinotHelixSegmentOnlineOfflineStateModelGenerator.PINOT_SEGMENT_ONLINE_OFFLINE_STATE_MODEL);
+ idealState.setRebalanceMode(IdealState.RebalanceMode.CUSTOMIZED);
+ idealState.setReplicas("1");
+ idealState.setNumPartitions(1);
+ idealState.setPartitionState(segmentInIdealState, "Server_unknown_0",
Helix.StateModel.SegmentStateModel.ONLINE);
+ _helixResourceManager.getHelixAdmin().addResource(getHelixClusterName(),
REMOVAL_TEST_TABLE, idealState);
+ for (String segmentName : segmentsWithZKMetadata) {
+ SegmentZKMetadata segmentZKMetadata = new SegmentZKMetadata(segmentName);
+ segmentZKMetadata.setStatus(Segment.Realtime.Status.IN_PROGRESS);
+ assertTrue(ZKMetadataProvider.setSegmentZKMetadata(_propertyStore,
REMOVAL_TEST_TABLE, segmentZKMetadata));
+ }
+ }
+
+ private void tearDownRemovalTestTable(List<String> segmentNames) {
+ _helixResourceManager.getHelixAdmin().dropResource(getHelixClusterName(),
REMOVAL_TEST_TABLE);
+ for (String segmentName : segmentNames) {
+ ZKMetadataProvider.removeSegmentZKMetadata(_propertyStore,
REMOVAL_TEST_TABLE, segmentName);
+ }
+ }
+
+ /// Covers
[PinotHelixResourceManager#removeInProgressSegmentsZKMetadataNotInIdealState]:
removes the ZK metadata of
+ /// the IN_PROGRESS segments that are not in the IdealState, and keeps the
others. The removal writes the unchanged
+ /// IdealState, so an IdealState update that read the IdealState before
fails.
+ @Test
+ public void testRemoveInProgressSegmentsZKMetadataNotInIdealState() {
+ String segmentInIdealState = "segmentInIdealState";
+ String segmentNotInIdealState = "segmentNotInIdealState";
+ String doneSegmentNotInIdealState = "doneSegmentNotInIdealState";
+ List<String> segments = List.of(segmentInIdealState,
segmentNotInIdealState, doneSegmentNotInIdealState);
+ setUpRemovalTestTable(segmentInIdealState, segments);
+ try {
+ SegmentZKMetadata doneSegmentZKMetadata = new
SegmentZKMetadata(doneSegmentNotInIdealState);
+ doneSegmentZKMetadata.setStatus(Segment.Realtime.Status.DONE);
+ assertTrue(ZKMetadataProvider.setSegmentZKMetadata(_propertyStore,
REMOVAL_TEST_TABLE, doneSegmentZKMetadata));
+ IdealState idealStateBefore =
_helixResourceManager.getTableIdealState(REMOVAL_TEST_TABLE);
+
+ List<String> segmentsToRemove = new ArrayList<>(segments);
+ segmentsToRemove.add("segmentWithoutZKMetadata");
+
assertEquals(_helixResourceManager.removeInProgressSegmentsZKMetadataNotInIdealState(REMOVAL_TEST_TABLE,
+ segmentsToRemove), Set.of(segmentNotInIdealState));
+ assertNotNull(ZKMetadataProvider.getSegmentZKMetadata(_propertyStore,
REMOVAL_TEST_TABLE, segmentInIdealState));
+ assertNull(ZKMetadataProvider.getSegmentZKMetadata(_propertyStore,
REMOVAL_TEST_TABLE, segmentNotInIdealState));
+ assertNotNull(
+ ZKMetadataProvider.getSegmentZKMetadata(_propertyStore,
REMOVAL_TEST_TABLE, doneSegmentNotInIdealState));
+ IdealState idealStateAfter =
_helixResourceManager.getTableIdealState(REMOVAL_TEST_TABLE);
+ assertEquals(idealStateAfter.getRecord().getMapFields(),
idealStateBefore.getRecord().getMapFields());
+ String idealStatePath =
PropertyPathBuilder.idealState(getHelixClusterName(), REMOVAL_TEST_TABLE);
+ // An IdealState update based on the IdealState read before the removal
fails
+ assertThrows(ZkBadVersionException.class,
+ () ->
_helixResourceManager.getHelixZkManager().getHelixDataAccessor().getBaseDataAccessor()
+ .set(idealStatePath, idealStateBefore.getRecord(),
idealStateBefore.getRecord().getVersion(),
+ AccessOption.PERSISTENT));
+
+ // The table was deleted
+
assertEquals(_helixResourceManager.removeInProgressSegmentsZKMetadataNotInIdealState("deletedTable_REALTIME",
+ segments), Set.of());
+
+ // Nothing to remove: the IdealState is not written
+
assertEquals(_helixResourceManager.removeInProgressSegmentsZKMetadataNotInIdealState(REMOVAL_TEST_TABLE,
+ segmentsToRemove), Set.of());
+
assertEquals(_helixResourceManager.getTableIdealState(REMOVAL_TEST_TABLE).getRecord().getVersion(),
+ idealStateAfter.getRecord().getVersion());
+ } finally {
+ tearDownRemovalTestTable(segments);
+ }
+ }
+
+ /// Another controller adds the segment to the IdealState after the removal
read the IdealState, e.g.
+ /// RealtimeSegmentValidationManager repairing the partition with it. The
removal must keep its ZK metadata.
+ @Test
+ public void
testRemoveInProgressSegmentsZKMetadataKeepsSegmentAddedToIdealStateConcurrently()
{
+ String segmentInIdealState = "segmentInIdealState";
+ String segmentToRemove = "segmentToRemove";
+ setUpRemovalTestTable(segmentInIdealState, List.of(segmentToRemove));
+ try {
+ AtomicBoolean added = new AtomicBoolean();
+ Runnable addToIdealState = () -> {
+ if (added.compareAndSet(false, true)) {
+ IdealState idealState =
_helixResourceManager.getTableIdealState(REMOVAL_TEST_TABLE);
+ idealState.setPartitionState(segmentToRemove, "Server_unknown_0",
+ Helix.StateModel.SegmentStateModel.CONSUMING);
+ _helixResourceManager.getHelixAdmin()
+ .setResourceIdealState(getHelixClusterName(),
REMOVAL_TEST_TABLE, idealState);
+ }
+ };
+
+
assertEquals(_helixResourceManager.removeInProgressSegmentsZKMetadataNotInIdealState(REMOVAL_TEST_TABLE,
+ List.of(segmentToRemove), addToIdealState), Set.of());
+ assertTrue(added.get());
+ assertNotNull(ZKMetadataProvider.getSegmentZKMetadata(_propertyStore,
REMOVAL_TEST_TABLE, segmentToRemove));
+
assertNotNull(_helixResourceManager.getTableIdealState(REMOVAL_TEST_TABLE).getInstanceStateMap(segmentToRemove));
+ } finally {
+ tearDownRemovalTestTable(List.of(segmentToRemove));
+ }
+ }
+
+ /// The segment ZK metadata is overwritten after the removal read it, e.g.
by an upload of a segment with the same
+ /// name. The removal must keep the new ZK metadata.
+ @Test
+ public void
testRemoveInProgressSegmentsZKMetadataKeepsSegmentZKMetadataReplacedConcurrently()
{
+ String segmentInIdealState = "segmentInIdealState";
+ String segmentToRemove = "segmentToRemove";
+ setUpRemovalTestTable(segmentInIdealState, List.of(segmentToRemove));
+ try {
+ AtomicBoolean replaced = new AtomicBoolean();
+ Runnable replaceSegmentZKMetadata = () -> {
+ if (replaced.compareAndSet(false, true)) {
+ SegmentZKMetadata uploadedSegmentZKMetadata = new
SegmentZKMetadata(segmentToRemove);
+
uploadedSegmentZKMetadata.setStatus(Segment.Realtime.Status.UPLOADED);
+ assertTrue(
+ ZKMetadataProvider.setSegmentZKMetadata(_propertyStore,
REMOVAL_TEST_TABLE, uploadedSegmentZKMetadata));
+ }
+ };
+
+
assertEquals(_helixResourceManager.removeInProgressSegmentsZKMetadataNotInIdealState(REMOVAL_TEST_TABLE,
+ List.of(segmentToRemove), replaceSegmentZKMetadata), Set.of());
+ assertTrue(replaced.get());
+ assertEquals(ZKMetadataProvider.getSegmentZKMetadata(_propertyStore,
REMOVAL_TEST_TABLE, segmentToRemove)
+ .getStatus(), Segment.Realtime.Status.UPLOADED);
+ } finally {
+ tearDownRemovalTestTable(List.of(segmentToRemove));
+ }
+ }
+
@AfterClass
public void tearDown() {
stopFakeInstances();
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 9216dddf860..0af3d8d542c 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
@@ -1988,6 +1988,154 @@ public class PinotLLCRealtimeSegmentManagerTest {
assertEquals(segmentManager._segmentZKMetadataMap.get(committingSegment).getStatus(),
Status.COMMITTING);
}
+ private static void setPauseState(IdealState idealState, boolean
tablePaused, List<Integer> pausedTopics) {
+ PauseState pauseState = new PauseState(tablePaused,
PauseState.ReasonCode.ADMINISTRATIVE, null,
+ Long.toString(CURRENT_TIME_MS), pausedTopics);
+
idealState.getRecord().setSimpleField(PinotLLCRealtimeSegmentManager.PAUSE_STATE,
pauseState.toJsonString());
+ }
+
+ /// The table is paused after a failed IdealState update kept the new
segment of the commit start. The retry
+ /// completes the commit without a new segment and removes the kept one, so
that resuming the table starts
+ /// consumption from the offset selected by the offset criteria instead of
adding the kept segment.
+ @Test
+ public void
testCommitStartRetryWhileTablePausedRemovesNewSegmentOfEarlierAttempt() {
+ FakePinotLLCRealtimeSegmentManager segmentManager =
setUpTableForCommitStart();
+ String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0,
CURRENT_TIME_MS).getSegmentName();
+ String newConsumingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 1,
CURRENT_TIME_MS).getSegmentName();
+ CommittingSegmentDescriptor committingSegmentDescriptor =
createCommittingSegmentDescriptor(committingSegment);
+ segmentManager._numFailedIdealStateUpdates = 1;
+ assertThrows(RuntimeException.class,
+ () ->
segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
committingSegmentDescriptor));
+
assertTrue(segmentManager._segmentZKMetadataMap.containsKey(newConsumingSegment));
+
+ setPauseState(segmentManager._idealState, true, List.of());
+ segmentManager._currentTimeMs += TimeUnit.MINUTES.toMillis(2);
+ segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
committingSegmentDescriptor);
+
assertFalse(segmentManager._segmentZKMetadataMap.containsKey(newConsumingSegment));
+ Map<String, Map<String, String>> instanceStatesMap =
segmentManager._idealState.getRecord().getMapFields();
+ assertFalse(instanceStatesMap.containsKey(newConsumingSegment));
+ assertEquals(new
HashSet<>(instanceStatesMap.get(committingSegment).values()),
Set.of(SegmentStateModel.ONLINE));
+
+ setPauseState(segmentManager._idealState, false, List.of());
+ segmentManager._exceededMaxSegmentCompletionTime = true;
+ segmentManager.ensureAllPartitionsConsuming(segmentManager._tableConfig,
segmentManager._streamConfigs,
+ segmentManager._idealState,
+ segmentManager.getNewStreamMetadataList(segmentManager._streamConfigs,
List.of(), segmentManager._idealState),
+ OffsetCriteria.LARGEST_OFFSET_CRITERIA);
+ String resumedSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 1,
segmentManager._currentTimeMs).getSegmentName();
+ assertFalse(instanceStatesMap.containsKey(newConsumingSegment));
+ assertEquals(new
HashSet<>(instanceStatesMap.get(resumedSegment).values()),
Set.of(SegmentStateModel.CONSUMING));
+
assertEquals(segmentManager._segmentZKMetadataMap.get(resumedSegment).getStartOffset(),
+ PARTITION_OFFSET.toString());
+ }
+
+ /// Like above, with 2 kept new segments, e.g. created by attempts of the
commit start on 2 controllers around a lead
+ /// change. The retry removes both.
+ @Test
+ public void
testCommitStartRetryWhileTablePausedRemovesNewSegmentsOfEarlierAttempts() {
+ FakePinotLLCRealtimeSegmentManager segmentManager =
setUpTableForCommitStart();
+ String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0,
CURRENT_TIME_MS).getSegmentName();
+ String newConsumingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 1,
CURRENT_TIME_MS).getSegmentName();
+ String otherNewConsumingSegment =
+ new LLCSegmentName(RAW_TABLE_NAME, 0, 1, CURRENT_TIME_MS +
TimeUnit.MINUTES.toMillis(1)).getSegmentName();
+ CommittingSegmentDescriptor committingSegmentDescriptor =
createCommittingSegmentDescriptor(committingSegment);
+ // Another attempt of the commit start creates its own new segment, and
its IdealState update fails as well
+ segmentManager._beforeCommitIdealStateUpdate = () -> {
+ segmentManager.persistSegmentZKMetadata(REALTIME_TABLE_NAME, new
SegmentZKMetadata(
+ new
ZNRecord(segmentManager._segmentZKMetadataMap.get(newConsumingSegment).toZNRecord(),
+ otherNewConsumingSegment)), -1);
+ segmentManager._beforeCommitIdealStateUpdate = () -> {
+ };
+ };
+ segmentManager._numFailedIdealStateUpdates = 1;
+ assertThrows(RuntimeException.class,
+ () ->
segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
committingSegmentDescriptor));
+
assertTrue(segmentManager._segmentZKMetadataMap.containsKey(newConsumingSegment));
+
assertTrue(segmentManager._segmentZKMetadataMap.containsKey(otherNewConsumingSegment));
+
+ setPauseState(segmentManager._idealState, true, List.of());
+ segmentManager._currentTimeMs += TimeUnit.MINUTES.toMillis(2);
+ segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
committingSegmentDescriptor);
+
assertFalse(segmentManager._segmentZKMetadataMap.containsKey(newConsumingSegment));
+
assertFalse(segmentManager._segmentZKMetadataMap.containsKey(otherNewConsumingSegment));
+ Map<String, Map<String, String>> instanceStatesMap =
segmentManager._idealState.getRecord().getMapFields();
+ assertEquals(new
HashSet<>(instanceStatesMap.get(committingSegment).values()),
Set.of(SegmentStateModel.ONLINE));
+ }
+
+ /// The retry of a commit start, which completes the commit without the new
segment kept by a failed attempt, stalls
+ /// before removing the kept segment. Meanwhile the table is resumed, and
validation adds the kept segment to the
+ /// IdealState. The retry must then keep its ZK metadata.
+ @Test
+ public void
testCommitStartRetryKeepsNewSegmentOfEarlierAttemptAddedByValidation() {
+ FakePinotLLCRealtimeSegmentManager segmentManager =
setUpTableForCommitStart();
+ String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0,
CURRENT_TIME_MS).getSegmentName();
+ String newConsumingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 1,
CURRENT_TIME_MS).getSegmentName();
+ CommittingSegmentDescriptor committingSegmentDescriptor =
createCommittingSegmentDescriptor(committingSegment);
+ segmentManager._numFailedIdealStateUpdates = 1;
+ assertThrows(RuntimeException.class,
+ () ->
segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
committingSegmentDescriptor));
+
+ setPauseState(segmentManager._idealState, true, List.of());
+ segmentManager._beforeRemovingUnusedNewSegments = () -> {
+ setPauseState(segmentManager._idealState, false, List.of());
+ segmentManager._exceededMaxSegmentCompletionTime = true;
+ segmentManager.ensureAllPartitionsConsuming();
+ };
+ segmentManager._currentTimeMs += TimeUnit.MINUTES.toMillis(2);
+ segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
committingSegmentDescriptor);
+ Map<String, Map<String, String>> instanceStatesMap =
segmentManager._idealState.getRecord().getMapFields();
+ assertEquals(new
HashSet<>(instanceStatesMap.get(newConsumingSegment).values()),
+ Set.of(SegmentStateModel.CONSUMING));
+
assertTrue(segmentManager._segmentZKMetadataMap.containsKey(newConsumingSegment));
+ }
+
+ /// The topic is paused after a failed IdealState update kept the new
segment of the commit start. The retry completes
+ /// the commit without a new segment and removes the kept one.
+ @Test
+ public void
testCommitStartRetryWhileTopicPausedRemovesNewSegmentOfEarlierAttempt() {
+ FakePinotLLCRealtimeSegmentManager segmentManager =
setUpTableForCommitStart();
+ String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0,
CURRENT_TIME_MS).getSegmentName();
+ String newConsumingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 1,
CURRENT_TIME_MS).getSegmentName();
+ CommittingSegmentDescriptor committingSegmentDescriptor =
createCommittingSegmentDescriptor(committingSegment);
+ segmentManager._numFailedIdealStateUpdates = 1;
+ assertThrows(RuntimeException.class,
+ () ->
segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
committingSegmentDescriptor));
+
assertTrue(segmentManager._segmentZKMetadataMap.containsKey(newConsumingSegment));
+
+ setPauseState(segmentManager._idealState, false, List.of(0));
+ segmentManager._currentTimeMs += TimeUnit.MINUTES.toMillis(2);
+ segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
committingSegmentDescriptor);
+
assertFalse(segmentManager._segmentZKMetadataMap.containsKey(newConsumingSegment));
+ Map<String, Map<String, String>> instanceStatesMap =
segmentManager._idealState.getRecord().getMapFields();
+ assertFalse(instanceStatesMap.containsKey(newConsumingSegment));
+ assertEquals(new
HashSet<>(instanceStatesMap.get(committingSegment).values()),
Set.of(SegmentStateModel.ONLINE));
+ }
+
+ /// The partition is removed from the stream after a failed IdealState
update kept the new segment of the commit
+ /// start. The retry completes the commit without adding the kept segment,
and removes it.
+ @Test
+ public void
testCommitStartRetryAfterPartitionRemovedFromStreamRemovesNewSegmentOfEarlierAttempt()
{
+ FakePinotLLCRealtimeSegmentManager segmentManager =
setUpTableForCommitStart();
+ String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0,
CURRENT_TIME_MS).getSegmentName();
+ String newConsumingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 1,
CURRENT_TIME_MS).getSegmentName();
+ CommittingSegmentDescriptor committingSegmentDescriptor =
createCommittingSegmentDescriptor(committingSegment);
+ segmentManager._numFailedIdealStateUpdates = 1;
+ assertThrows(RuntimeException.class,
+ () ->
segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
committingSegmentDescriptor));
+
assertTrue(segmentManager._segmentZKMetadataMap.containsKey(newConsumingSegment));
+
+ segmentManager._streamMetadataList = List.of(new
StreamMetadata(segmentManager._streamConfigs.get(0), 3,
+ IntStream.range(1, 4)
+ .mapToObj(partition -> new PartitionGroupMetadata(partition,
PARTITION_OFFSET))
+ .collect(Collectors.toList())));
+ segmentManager._currentTimeMs += TimeUnit.MINUTES.toMillis(2);
+ segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
committingSegmentDescriptor);
+
assertFalse(segmentManager._segmentZKMetadataMap.containsKey(newConsumingSegment));
+ Map<String, Map<String, String>> instanceStatesMap =
segmentManager._idealState.getRecord().getMapFields();
+ assertFalse(instanceStatesMap.containsKey(newConsumingSegment));
+ assertEquals(new
HashSet<>(instanceStatesMap.get(committingSegment).values()),
Set.of(SegmentStateModel.ONLINE));
+ }
+
/// Runs the production IdealState update with a write that is applied but
reported as failed, for both the group
/// commit and the single commit.
@Test
@@ -3290,6 +3438,9 @@ public class PinotLLCRealtimeSegmentManagerTest {
// Runs before updating the IdealState (step 3) of a segment commit
Runnable _beforeCommitIdealStateUpdate = () -> {
};
+ // Runs before removing the ZK metadata of the new segments that a segment
commit did not add to the IdealState
+ Runnable _beforeRemovingUnusedNewSegments = () -> {
+ };
FileUploadDownloadClient _mockedFileUploadDownloadClient;
PinotHelixResourceManager _mockResourceManager;
// Segment name to the server endpoint re-ingestion was triggered on
@@ -3466,6 +3617,21 @@ public class PinotLLCRealtimeSegmentManagerTest {
_beforeCommitIdealStateUpdate.run();
}
+ /// Like the production removal, which is fenced against concurrent
updates, only removes the segments that are
+ /// IN_PROGRESS and not in the IdealState when they are removed.
+ @Override
+ void removeUnusedNewSegmentsZKMetadata(String realtimeTableName,
Set<String> segmentNames) {
+ _beforeRemovingUnusedNewSegments.run();
+ for (String segmentName : segmentNames) {
+ SegmentZKMetadata segmentZKMetadata =
_segmentZKMetadataMap.get(segmentName);
+ if (segmentZKMetadata != null && segmentZKMetadata.getStatus() ==
Status.IN_PROGRESS
+ &&
!_idealState.getRecord().getMapFields().containsKey(segmentName)) {
+ _segmentZKMetadataMap.remove(segmentName);
+ _segmentZKMetadataVersionMap.remove(segmentName);
+ }
+ }
+ }
+
@Override
public IdealState getIdealState(String realtimeTableName) {
return _idealState;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
