You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ra...@apache.org on 2026/10/20 23:28:40 UTC
(pinot) branch master updated: Resume a pauseless commit start after its IdealState update fails (#19740)
This is an automated email from the ASF dual-hosted git repository.
raghavyadav01 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 89b84bad2e2 Resume a pauseless commit start after its IdealState update fails (#19740)
89b84bad2e2 is described below
commit 89b84bad2e2fcefcb7dac4ec0bbdb148fdb0e023
Author: Suvodeep Pyne <su...@users.noreply.github.com>
AuthorDate: Tue Oct 20 16:28:33 2026 -0700
Resume a pauseless commit start after its IdealState update fails (#19740)
* Resume a pauseless commit start after its IdealState update fails
The pauseless commit start moves the committing segment to COMMITTING in ZK
(step 1), creates the next segment's ZK metadata (step 2), then updates the
IdealState (step 3). When step 3 fails, e.g. after losing every retry to
IdealState version conflicts, the segment stays COMMITTING, and every retry
of the commit failed with "should be IN_PROGRESS, found: COMMITTING" until
RealtimeSegmentValidationManager repaired the partition after
max.segment.completion.time.millis. The same happened when the controller
stopped between step 1 and step 3.
A commit start for a segment that is already COMMITTING with the same end
offset now resumes the earlier attempt from step 2, until the max segment
completion time, after which the validation manager still owns the repair.
Steps 2 and 3 are made idempotent so that the attempts of a commit start,
including attempts on different controllers around a lead change, converge:
- A resumed attempt uses the new segment created by an earlier attempt, if
any. The new segment is named after its creation time with minute
granularity, and is created between step 1 and the max segment completion
time, so the resumed attempt probes the names of the minutes in between.
Otherwise it creates the new segment with the current time, as the first
attempt does, so the segment is never named after an older time.
- New segment ZK metadata is created atomically only if absent and is never
overwritten, since the segment might be consuming or committed already.
- The IdealState updater skips an update that is already applied (all
replicas of the committing segment ONLINE and the new segment present).
Previously a retry after a write that succeeded but was reported as failed
failed every remaining attempt, and every update batched with it, on
"Failed to find instance in CONSUMING state".
- When step 3 fails, the controller re-reads the IdealState and proceeds if
the update was applied. Otherwise it keeps the new segment's ZK metadata for
the next attempt, or for the validation manager, which adds such a segment
to the IdealState. It removes the metadata only when the new segment is not
in the IdealState and the committing segment is no longer CONSUMING, the
only state in which no attempt can add the segment anymore.
Attempts on two controllers, or a kept new segment and a validation repair
that started before it was created, can still create two new segments with
the same sequence number. When several segments of a partition have the
highest sequence number, the validation manager now takes the one in the
IdealState as the latest segment, instead of whichever is listed first, which
could make it fail every run with "Segment ... is a duplicate of existing
segment ...".
No ZK metadata is rolled back, so the IdealState version check of step 3
remains the only fence against a concurrent repair by the validation manager.
* Remove a commit start test that duplicates another one
testCommitStartFailureKeepsNewSegmentAddedConcurrently exercised the same
branch as testCommitStartSucceedsWhenFailedIdealStateUpdateWasApplied: after
a failed IdealState update, the controller re-reads the IdealState, finds the
completion applied and proceeds. The failure handler does not depend on
whether the attempt was resumed.
* Document assumptions of the commit start resume
Note that probing for the new segment of an earlier attempt relies on the committing segment ZK metadata not being rewritten while it stays COMMITTING, and which segment is the latest one when several segments with the highest sequence number are missing from the IdealState.
---
.../realtime/BlockingSegmentCompletionFSM.java | 5 +-
.../realtime/PinotLLCRealtimeSegmentManager.java | 252 +++++++++--
.../PinotLLCRealtimeSegmentManagerTest.java | 460 ++++++++++++++++++++-
3 files changed, 686 insertions(+), 31 deletions(-)
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/BlockingSegmentCompletionFSM.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/BlockingSegmentCompletionFSM.java
index 381bebfdb30..c8c4df32241 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/BlockingSegmentCompletionFSM.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/BlockingSegmentCompletionFSM.java
@@ -141,8 +141,9 @@ public class BlockingSegmentCompletionFSM implements SegmentCompletionFSM {
_maxTimeAllowedToCommitMs = _startTimeMs + _initialCommitTimeMs;
_controllerVipUrl = segmentCompletionManager.getControllerVipUrl();
- // NOTE: If segment ZK status is COMMITTING, The current behaviour expects the segment protocol calls to fail and
- // abort leaving it to realtime segment validation job to fix it.
+ // NOTE: If segment ZK status is COMMITTING, a pauseless commit start with the same end offset resumes the commit
+ // (see PinotLLCRealtimeSegmentManager). Otherwise the segment protocol calls fail and abort, leaving it to realtime
+ // segment validation job to fix it.
if (segmentMetadata.getStatus().isCompleted()) {
_state = BlockingSegmentCompletionFSMState.COMMITTED;
String endOffset = segmentMetadata.getEndOffset();
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 9303c6e5bcb..b4add6ac69e 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
@@ -659,9 +659,15 @@ public class PinotLLCRealtimeSegmentManager implements PinotClusterConfigChangeL
// Step-1: Update PROPERTYSTORE
LOGGER.info("Committing segment metadata for segment: {}", committingSegmentName);
long startTimeNs1 = System.nanoTime();
- SegmentZKMetadata committingSegmentZKMetadata = toCommitting
- ? updateSegmentZKMetadataToCommitting(realtimeTableName, committingSegmentDescriptor)
- : updateSegmentZKMetadataToDone(realtimeTableName, committingSegmentDescriptor, Status.IN_PROGRESS);
+ boolean resumed = toCommitting && isCommitStartResumable(realtimeTableName, committingSegmentDescriptor);
+ SegmentZKMetadata committingSegmentZKMetadata;
+ if (resumed) {
+ committingSegmentZKMetadata = getSegmentZKMetadata(realtimeTableName, committingSegmentName);
+ } else {
+ committingSegmentZKMetadata = toCommitting
+ ? updateSegmentZKMetadataToCommitting(realtimeTableName, committingSegmentDescriptor)
+ : updateSegmentZKMetadataToDone(realtimeTableName, committingSegmentDescriptor, Status.IN_PROGRESS);
+ }
preProcessNewSegmentZKMetadata();
@@ -669,7 +675,7 @@ public class PinotLLCRealtimeSegmentManager implements PinotClusterConfigChangeL
long startTimeNs2 = System.nanoTime();
String newConsumingSegmentName =
createNewSegmentMetadata(tableConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
- instancePartitions);
+ instancePartitions, toCommitting, resumed);
preProcessCommitIdealStateUpdate();
@@ -677,33 +683,41 @@ public class PinotLLCRealtimeSegmentManager implements PinotClusterConfigChangeL
LOGGER.info("Updating Idealstate for previous: {} and new segment: {}", committingSegmentName,
newConsumingSegmentName);
long startTimeNs3 = System.nanoTime();
- Map<String, Map<String, String>> instanceStatesMapAfterStep3;
- boolean newConsumingSegmentInIdealState = false;
+ IdealState idealState;
// When multiple segments of the same table complete around the same time it is possible that
// the idealstate update fails due to contention. We serialize the updates to the idealstate
// to reduce this contention. We may still contend with RetentionManager, or other updates
// to idealstate from other controllers, but then we have the retry mechanism to get around that.
try {
- IdealState idealState =
+ idealState =
updateIdealStateForSegments(tableConfig, committingSegmentName, newConsumingSegmentName, instancePartitions);
- instanceStatesMapAfterStep3 = idealState.getRecord().getMapFields();
- 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);
+ } catch (RuntimeException e) {
+ if (!toCommitting) {
+ if (newConsumingSegmentName != null) {
removeSegmentZKMetadataBestEffort(realtimeTableName, newConsumingSegmentName);
- newConsumingSegmentName = null;
}
+ throw e;
}
- } catch (RuntimeException e) {
- if (newConsumingSegmentName != null) {
+ idealState =
+ handleCommitStartIdealStateUpdateFailure(realtimeTableName, committingSegmentName, newConsumingSegmentName);
+ if (idealState == null) {
+ throw e;
+ }
+ }
+
+ Map<String, Map<String, String>> instanceStatesMapAfterStep3 = idealState.getRecord().getMapFields();
+ boolean newConsumingSegmentInIdealState = false;
+ 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;
}
- throw e;
}
long endTimeNs = System.nanoTime();
@@ -806,11 +820,65 @@ public class PinotLLCRealtimeSegmentManager implements PinotClusterConfigChangeL
return segmentZKMetadata;
}
+ /// Returns `true` if a pauseless commit start can resume an earlier attempt that moved the segment to COMMITTING
+ /// with the same end offset, but did not complete, e.g. because the IdealState update failed or the controller
+ /// stopped. Steps 2 and 3 are idempotent for the same segment and end offset: the resumed attempt uses the new
+ /// segment created by an earlier attempt if any (see findNewSegmentOfEarlierAttempt), new segment ZK metadata is only
+ /// created if absent, and the IdealState update is skipped when it is already applied. Once the max segment
+ /// completion time is exceeded, RealtimeSegmentValidationManager repairs the segment instead.
+ private boolean isCommitStartResumable(String realtimeTableName,
+ CommittingSegmentDescriptor committingSegmentDescriptor) {
+ String segmentName = committingSegmentDescriptor.getSegmentName();
+ SegmentZKMetadata segmentZKMetadata = getSegmentZKMetadata(realtimeTableName, segmentName);
+ if (segmentZKMetadata.getStatus() != Status.COMMITTING) {
+ return false;
+ }
+ String endOffset = segmentZKMetadata.getEndOffset();
+ if (!committingSegmentDescriptor.getNextOffset().equals(endOffset)) {
+ LOGGER.warn("Cannot resume commit start for segment: {} with end offset: {}, segment is COMMITTING with end "
+ + "offset: {}", segmentName, committingSegmentDescriptor.getNextOffset(), endOffset);
+ return false;
+ }
+ if (isExceededMaxSegmentCompletionTime(realtimeTableName, segmentName, getCurrentTimeMs())) {
+ LOGGER.warn("Cannot resume commit start for segment: {} after the max segment completion time", segmentName);
+ return false;
+ }
+ LOGGER.info("Resuming commit start for segment: {} which is COMMITTING with end offset: {}", segmentName,
+ endOffset);
+ return true;
+ }
+
+ /// Returns the new segment created by an earlier attempt of a resumed commit start, or `null` if there is none.
+ ///
+ /// 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
+ /// segment completion time. Probes the names for every minute in between, with one minute of slack on both ends for
+ /// 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) {
+ 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;
+ 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;
+ }
+ }
+ return null;
+ }
+
// Step 2: Create new segment metadata
@Nullable
private String createNewSegmentMetadata(TableConfig tableConfig,
CommittingSegmentDescriptor committingSegmentDescriptor, SegmentZKMetadata committingSegmentZKMetadata,
- InstancePartitions instancePartitions) {
+ InstancePartitions instancePartitions, boolean createIfAbsent, boolean resumed) {
String committingSegmentName = committingSegmentDescriptor.getSegmentName();
String realtimeTableName = tableConfig.getTableName();
@@ -836,17 +904,35 @@ 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;
+ }
+ }
String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
long newSegmentCreationTimeMs = getCurrentTimeMs();
LLCSegmentName newLLCSegment = new LLCSegmentName(rawTableName, committingSegmentPartitionGroupId,
committingLLCSegment.getSequenceNumber() + 1, newSegmentCreationTimeMs);
+ newConsumingSegmentName = newLLCSegment.getSegmentName();
StreamConfig streamConfig =
IngestionConfigUtils.getStreamConfigFromPinotPartitionId(streamConfigs, committingSegmentPartitionGroupId);
- createNewSegmentZKMetadata(tableConfig, streamConfig, newLLCSegment, newSegmentCreationTimeMs,
- committingSegmentDescriptor, committingSegmentZKMetadata, instancePartitions, partitionIds.size(),
- numReplicas);
- newConsumingSegmentName = newLLCSegment.getSegmentName();
+ SegmentZKMetadata newSegmentZKMetadata =
+ buildNewSegmentZKMetadata(tableConfig, streamConfig, newLLCSegment, newSegmentCreationTimeMs,
+ committingSegmentDescriptor, committingSegmentZKMetadata, instancePartitions, partitionIds.size(),
+ numReplicas);
+ if (createIfAbsent) {
+ if (!createSegmentZKMetadataIfAbsent(realtimeTableName, newSegmentZKMetadata)) {
+ LOGGER.info("Using existing segment metadata for new segment: {}", newConsumingSegmentName);
+ return newConsumingSegmentName;
+ }
+ } else {
+ persistSegmentZKMetadata(realtimeTableName, newSegmentZKMetadata, -1);
+ }
LOGGER.info("Created new segment metadata for segment: {} with status: {}.", newConsumingSegmentName,
Status.IN_PROGRESS);
} else {
@@ -873,6 +959,73 @@ public class PinotLLCRealtimeSegmentManager implements PinotClusterConfigChangeL
}
}
+ @VisibleForTesting
+ boolean segmentZKMetadataExists(String realtimeTableName, String segmentName) {
+ String segmentMetadataPath =
+ ZKMetadataProvider.constructPropertyStorePathForSegment(realtimeTableName, segmentName);
+ return _propertyStore.exists(segmentMetadataPath, AccessOption.PERSISTENT);
+ }
+
+ /// Atomically creates the segment ZK metadata. Returns `false` if it already exists.
+ @VisibleForTesting
+ boolean createSegmentZKMetadataIfAbsent(String realtimeTableName, SegmentZKMetadata segmentZKMetadata) {
+ String segmentName = segmentZKMetadata.getSegmentName();
+ String segmentMetadataPath =
+ ZKMetadataProvider.constructPropertyStorePathForSegment(realtimeTableName, segmentName);
+ LOGGER.info("Creating segment ZK metadata for segment: {}", segmentName);
+ try {
+ if (_propertyStore.create(segmentMetadataPath, segmentZKMetadata.toZNRecord(), AccessOption.PERSISTENT)) {
+ return true;
+ }
+ Preconditions.checkState(_propertyStore.exists(segmentMetadataPath, AccessOption.PERSISTENT),
+ "Failed to create segment ZK metadata for segment: %s of table: %s", segmentName, realtimeTableName);
+ return false;
+ } catch (Exception e) {
+ _controllerMetrics.addMeteredTableValue(realtimeTableName, ControllerMeter.LLC_ZOOKEEPER_UPDATE_FAILURES, 1L);
+ throw e;
+ }
+ }
+
+ /// Handles a failed IdealState update (step 3) of a pauseless commit start. Re-reads the IdealState to find out
+ /// whether the update was applied (e.g. the write succeeded but the ZK connection was lost before the response
+ /// arrived). If it was, returns the IdealState so that the commit can proceed. Otherwise returns `null`, and the
+ /// committing segment stays COMMITTING so that the next attempt of the commit can resume it (see
+ /// isCommitStartResumable).
+ ///
+ /// The ZK metadata of the new consuming segment is kept for the next attempt, or for
+ /// RealtimeSegmentValidationManager to add the segment to the IdealState. It is only removed when the segment can no
+ /// longer be added, i.e. it is not in the IdealState and the committing segment is no longer CONSUMING (e.g. the
+ /// validation manager repaired the partition with another segment). Adding the segment requires the committing
+ /// segment to be CONSUMING, and the IdealState version check prevents a concurrent attempt from adding it based on
+ /// an older IdealState.
+ @Nullable
+ private IdealState handleCommitStartIdealStateUpdateFailure(String realtimeTableName, String committingSegmentName,
+ @Nullable String newConsumingSegmentName) {
+ IdealState idealState;
+ try {
+ idealState = getIdealState(realtimeTableName);
+ } catch (Exception e) {
+ LOGGER.warn("Failed to read IdealState of table: {} after failing to update it for segment: {}",
+ realtimeTableName, committingSegmentName, e);
+ return null;
+ }
+ if (isSegmentCompletionApplied(idealState, committingSegmentName, newConsumingSegmentName)) {
+ LOGGER.info("IdealState of table: {} already reflects the completion of segment: {}, proceeding with the commit",
+ realtimeTableName, committingSegmentName);
+ return idealState;
+ }
+ if (newConsumingSegmentName != null && idealState.getInstanceStateMap(newConsumingSegmentName) == null) {
+ Map<String, String> committingSegmentInstanceStateMap = idealState.getInstanceStateMap(committingSegmentName);
+ if (committingSegmentInstanceStateMap == null || !committingSegmentInstanceStateMap.containsValue(
+ SegmentStateModel.CONSUMING)) {
+ LOGGER.info("Removing segment ZK metadata for new segment: {} because segment: {} is no longer CONSUMING",
+ newConsumingSegmentName, committingSegmentName);
+ removeSegmentZKMetadataBestEffort(realtimeTableName, newConsumingSegmentName);
+ }
+ }
+ return null;
+ }
+
// Step 3: Update IdealState
private IdealState updateIdealStateForSegments(TableConfig tableConfig, String committingSegmentName,
String newConsumingSegmentName, InstancePartitions instancePartitions) {
@@ -977,6 +1130,17 @@ public class PinotLLCRealtimeSegmentManager implements PinotClusterConfigChangeL
LLCSegmentName newLLCSegmentName, long creationTimeMs, CommittingSegmentDescriptor committingSegmentDescriptor,
@Nullable SegmentZKMetadata committingSegmentZKMetadata, InstancePartitions instancePartitions, int numPartitions,
int numReplicas) {
+ SegmentZKMetadata newSegmentZKMetadata =
+ buildNewSegmentZKMetadata(tableConfig, streamConfig, newLLCSegmentName, creationTimeMs,
+ committingSegmentDescriptor, committingSegmentZKMetadata, instancePartitions, numPartitions, numReplicas);
+ persistSegmentZKMetadata(tableConfig.getTableName(), newSegmentZKMetadata, -1);
+ }
+
+ /// Builds segment ZK metadata for the new CONSUMING segment.
+ private SegmentZKMetadata buildNewSegmentZKMetadata(TableConfig tableConfig, StreamConfig streamConfig,
+ LLCSegmentName newLLCSegmentName, long creationTimeMs, CommittingSegmentDescriptor committingSegmentDescriptor,
+ @Nullable SegmentZKMetadata committingSegmentZKMetadata, InstancePartitions instancePartitions, int numPartitions,
+ int numReplicas) {
String realtimeTableName = tableConfig.getTableName();
String segmentName = newLLCSegmentName.getSegmentName();
@@ -1027,8 +1191,7 @@ public class PinotLLCRealtimeSegmentManager implements PinotClusterConfigChangeL
flushThresholdUpdater.updateFlushThreshold(streamConfig, newSegmentZKMetadata,
getMaxNumPartitionsPerInstance(instancePartitions, numPartitions, numReplicas));
updateFlushThresholdGauge(streamConfig, newSegmentZKMetadata.getSizeThresholdToFlushSegment());
-
- persistSegmentZKMetadata(realtimeTableName, newSegmentZKMetadata, -1);
+ return newSegmentZKMetadata;
}
private void updateCommittingSegmentSizeGauge(StreamConfig streamConfig, long segmentSize) {
@@ -1384,9 +1547,15 @@ 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.
+ ///
/// @param realtimeTableName Realtime table name
+ /// @param instanceStatesMap Instance states map of the IdealState
/// @return Map from partition group id to the latest LLC realtime segment ZK metadata
- private Map<Integer, SegmentZKMetadata> getLatestSegmentZKMetadataMap(String realtimeTableName) {
+ private Map<Integer, SegmentZKMetadata> getLatestSegmentZKMetadataMap(String realtimeTableName,
+ Map<String, Map<String, String>> instanceStatesMap) {
List<String> segments = getLLCSegments(realtimeTableName);
Map<Integer, LLCSegmentName> latestLLCSegmentNameMap = new HashMap<>();
@@ -1398,6 +1567,11 @@ public class PinotLLCRealtimeSegmentManager implements PinotClusterConfigChangeL
} else {
if (llcSegmentName.getSequenceNumber() > latestLLCSegmentName.getSequenceNumber()) {
return llcSegmentName;
+ } else if (llcSegmentName.getSequenceNumber() == latestLLCSegmentName.getSequenceNumber()) {
+ LOGGER.warn("Found segments: {} and {} with the same sequence number in partition: {} of table: {}",
+ latestLLCSegmentName.getSegmentName(), segmentName, partitionId, realtimeTableName);
+ return !instanceStatesMap.containsKey(latestLLCSegmentName.getSegmentName())
+ && instanceStatesMap.containsKey(segmentName) ? llcSegmentName : latestLLCSegmentName;
} else {
return latestLLCSegmentName;
}
@@ -1503,6 +1677,13 @@ public class PinotLLCRealtimeSegmentManager implements PinotClusterConfigChangeL
throw new HelixHelper.PermanentUpdaterException(
"Exceeded max segment completion time for segment " + committingSegmentName);
}
+ // A previous attempt might have been applied even though it was reported as failed, e.g. when the write
+ // succeeded but the ZK connection was lost before the response arrived
+ if (isSegmentCompletionApplied(idealState, committingSegmentName, newSegmentName)) {
+ LOGGER.info("IdealState of table: {} already reflects the completion of segment: {}, skipping the update",
+ realtimeTableName, committingSegmentName);
+ return idealState;
+ }
updateInstanceStatesForNewConsumingSegment(idealState.getRecord().getMapFields(), committingSegmentName,
isTablePaused(idealState) || isTopicPaused(idealState, committingSegmentName), newSegmentName,
segmentAssignment, instancePartitionsMap);
@@ -1517,6 +1698,20 @@ public class PinotLLCRealtimeSegmentManager implements PinotClusterConfigChangeL
false);
}
+ /// Returns `true` if the IdealState already reflects the completion of the committing segment, i.e. all its replicas
+ /// are ONLINE, and the new consuming segment (if one is expected) is present.
+ @VisibleForTesting
+ static boolean isSegmentCompletionApplied(IdealState idealState, String committingSegmentName,
+ @Nullable String newSegmentName) {
+ Map<String, String> committingSegmentInstanceStateMap = idealState.getInstanceStateMap(committingSegmentName);
+ if (committingSegmentInstanceStateMap == null || committingSegmentInstanceStateMap.isEmpty()
+ || !committingSegmentInstanceStateMap.values().stream().allMatch(SegmentStateModel.ONLINE::equals)) {
+ return false;
+ }
+ return newSegmentName == null || isTablePaused(idealState) || isTopicPaused(idealState, committingSegmentName)
+ || idealState.getInstanceStateMap(newSegmentName) != null;
+ }
+
public static boolean isTablePaused(IdealState idealState) {
PauseState pauseState = extractTablePauseState(idealState);
if (pauseState != null) {
@@ -1735,7 +1930,8 @@ public class PinotLLCRealtimeSegmentManager implements PinotClusterConfigChangeL
StreamConsumerFactoryProvider.create(streamConfigs.get(0)).createStreamMsgOffsetFactory();
// Get the latest segment ZK metadata for each partition
- Map<Integer, SegmentZKMetadata> latestSegmentZKMetadataMap = getLatestSegmentZKMetadataMap(realtimeTableName);
+ Map<Integer, SegmentZKMetadata> latestSegmentZKMetadataMap =
+ getLatestSegmentZKMetadataMap(realtimeTableName, instanceStatesMap);
// Create a map from partition id to start offset
// TODO: Directly return map from StreamMetadataProvider
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 34deec66251..2f943077ccb 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
@@ -30,6 +30,7 @@ import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -46,9 +47,12 @@ import javax.annotation.Nullable;
import javax.ws.rs.core.Response;
import org.apache.commons.io.FileUtils;
import org.apache.helix.AccessOption;
+import org.apache.helix.BaseDataAccessor;
import org.apache.helix.ClusterMessagingService;
import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
+import org.apache.helix.PropertyKey;
import org.apache.helix.model.ExternalView;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.InstanceConfig;
@@ -1486,6 +1490,403 @@ public class PinotLLCRealtimeSegmentManagerTest {
AccessOption.PERSISTENT);
}
+ /// Sets up a table for pauseless commit start tests: the committing segments list can be created, and removing a
+ /// segment ZK metadata removes it from the fake segment manager.
+ private FakePinotLLCRealtimeSegmentManager setUpTableForCommitStart() {
+ FakePinotLLCRealtimeSegmentManager segmentManager = new FakePinotLLCRealtimeSegmentManager();
+ setUpNewTable(segmentManager, 2, 5, 4);
+ ZkHelixPropertyStore<ZNRecord> propertyStore = segmentManager._mockResourceManager.getPropertyStore();
+ when(propertyStore.create(anyString(), any(), eq(AccessOption.PERSISTENT))).thenReturn(true);
+ when(propertyStore.remove(anyString(), eq(AccessOption.PERSISTENT))).thenAnswer(invocation -> {
+ String path = invocation.getArgument(0);
+ String segmentName = path.substring(path.lastIndexOf('/') + 1);
+ segmentManager._segmentZKMetadataMap.remove(segmentName);
+ segmentManager._segmentZKMetadataVersionMap.remove(segmentName);
+ return true;
+ });
+ return segmentManager;
+ }
+
+ /// Simulates another controller completing the commit start of the segment: the new segment is created (with the
+ /// given status), the committing segment becomes ONLINE and the new segment CONSUMING in the IdealState.
+ private static void completeCommitStartElsewhere(FakePinotLLCRealtimeSegmentManager segmentManager,
+ String committingSegment, String newSegment, Status newSegmentStatus) {
+ SegmentZKMetadata newSegmentZKMetadata = new SegmentZKMetadata(newSegment);
+ newSegmentZKMetadata.setStatus(newSegmentStatus);
+ newSegmentZKMetadata.setStartOffset(NEXT_OFFSET);
+ segmentManager.persistSegmentZKMetadata(REALTIME_TABLE_NAME, newSegmentZKMetadata, -1);
+ Map<String, Map<String, String>> instanceStatesMap = segmentManager._idealState.getRecord().getMapFields();
+ instanceStatesMap.put(newSegment, new TreeMap<>(instanceStatesMap.get(committingSegment)));
+ instanceStatesMap.get(committingSegment).replaceAll((instance, state) -> SegmentStateModel.ONLINE);
+ }
+
+ @Test
+ public void testCommitStartResumesAfterIdealStateUpdateFailure() {
+ 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));
+ // The committing segment stays COMMITTING, and the new segment is kept for the next attempt
+ SegmentZKMetadata committingSegmentZKMetadata = segmentManager._segmentZKMetadataMap.get(committingSegment);
+ assertEquals(committingSegmentZKMetadata.getStatus(), Status.COMMITTING);
+ assertEquals(committingSegmentZKMetadata.getEndOffset(), NEXT_OFFSET);
+ assertEquals(segmentManager._segmentZKMetadataMap.get(newConsumingSegment).getStatus(), Status.IN_PROGRESS);
+ int newSegmentZKMetadataVersion = segmentManager._segmentZKMetadataVersionMap.get(newConsumingSegment);
+ Map<String, Map<String, String>> instanceStatesMap = segmentManager._idealState.getRecord().getMapFields();
+ assertEquals(new HashSet<>(instanceStatesMap.get(committingSegment).values()),
+ Set.of(SegmentStateModel.CONSUMING));
+ assertFalse(instanceStatesMap.containsKey(newConsumingSegment));
+
+ // The retry resumes the commit start with the same new segment, although the current time moved on
+ segmentManager._currentTimeMs += TimeUnit.MINUTES.toMillis(2);
+ segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME, committingSegmentDescriptor);
+ committingSegmentZKMetadata = segmentManager._segmentZKMetadataMap.get(committingSegment);
+ assertEquals(committingSegmentZKMetadata.getStatus(), Status.COMMITTING);
+ assertEquals(committingSegmentZKMetadata.getEndOffset(), NEXT_OFFSET);
+ assertEquals(segmentManager._segmentZKMetadataMap.keySet().stream()
+ .filter(segment -> new LLCSegmentName(segment).getPartitionGroupId() == 0)
+ .collect(Collectors.toSet()), Set.of(committingSegment, newConsumingSegment));
+ assertEquals((int) segmentManager._segmentZKMetadataVersionMap.get(newConsumingSegment),
+ newSegmentZKMetadataVersion);
+ assertEquals(new HashSet<>(instanceStatesMap.get(committingSegment).values()), Set.of(SegmentStateModel.ONLINE));
+ assertEquals(new HashSet<>(instanceStatesMap.get(newConsumingSegment).values()),
+ Set.of(SegmentStateModel.CONSUMING));
+ }
+
+ @Test
+ public void testCommitStartResumeCreatesNewSegmentWithCurrentTimeWhenEarlierAttemptCreatedNone() {
+ FakePinotLLCRealtimeSegmentManager segmentManager = setUpTableForCommitStart();
+ String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(committingSegment);
+
+ // The first attempt fails after step 1, before creating the new segment
+ segmentManager._beforeNewSegmentZKMetadata = () -> {
+ throw new RuntimeException("Injected failure before creating the new segment");
+ };
+ assertThrows(RuntimeException.class,
+ () -> segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME, committingSegmentDescriptor));
+ segmentManager._beforeNewSegmentZKMetadata = () -> {
+ };
+
+ // The resumed attempt creates the new segment with its own creation time, but fails to update the IdealState
+ segmentManager._currentTimeMs += TimeUnit.MINUTES.toMillis(6);
+ long newSegmentCreationTimeMs = segmentManager._currentTimeMs;
+ String newConsumingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 1, newSegmentCreationTimeMs).getSegmentName();
+ segmentManager._numFailedIdealStateUpdates = 1;
+ assertThrows(RuntimeException.class,
+ () -> segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME, committingSegmentDescriptor));
+ assertEquals(segmentManager._segmentZKMetadataMap.get(newConsumingSegment).getCreationTime(),
+ newSegmentCreationTimeMs);
+
+ // The next resumed attempt finds that new segment
+ segmentManager._currentTimeMs += TimeUnit.MINUTES.toMillis(1);
+ segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME, committingSegmentDescriptor);
+ assertEquals(segmentManager._segmentZKMetadataMap.keySet().stream()
+ .filter(segment -> new LLCSegmentName(segment).getPartitionGroupId() == 0)
+ .collect(Collectors.toSet()), Set.of(committingSegment, newConsumingSegment));
+ assertEquals(new HashSet<>(segmentManager._idealState.getInstanceStateMap(newConsumingSegment).values()),
+ Set.of(SegmentStateModel.CONSUMING));
+ }
+
+ @Test
+ public void testCommitStartResumeFindsNewSegmentAtEdgeOfClockSkewAllowance() {
+ FakePinotLLCRealtimeSegmentManager segmentManager = setUpTableForCommitStart();
+ String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(committingSegment);
+ long minuteStartMs = CURRENT_TIME_MS / 60_000 * 60_000;
+
+ // Step 1 at second 59 of a minute, then the first attempt fails before creating the new segment
+ segmentManager._currentTimeMs = minuteStartMs + 59_000;
+ segmentManager._beforeNewSegmentZKMetadata = () -> {
+ throw new RuntimeException("Injected failure before creating the new segment");
+ };
+ assertThrows(RuntimeException.class,
+ () -> segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME, committingSegmentDescriptor));
+ segmentManager._beforeNewSegmentZKMetadata = () -> {
+ };
+
+ // Another controller, with a clock less than a minute ahead, created the new segment in the next minute
+ segmentManager._currentTimeMs = minuteStartMs + TimeUnit.MINUTES.toMillis(10) + 1_000;
+ String newConsumingSegment =
+ new LLCSegmentName(RAW_TABLE_NAME, 0, 1, minuteStartMs + TimeUnit.MINUTES.toMillis(11)).getSegmentName();
+ SegmentZKMetadata newSegmentZKMetadata = new SegmentZKMetadata(newConsumingSegment);
+ newSegmentZKMetadata.setStatus(Status.IN_PROGRESS);
+ newSegmentZKMetadata.setStartOffset(NEXT_OFFSET);
+ segmentManager.persistSegmentZKMetadata(REALTIME_TABLE_NAME, newSegmentZKMetadata, -1);
+
+ segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME, committingSegmentDescriptor);
+ assertEquals(segmentManager._segmentZKMetadataMap.keySet().stream()
+ .filter(segment -> new LLCSegmentName(segment).getPartitionGroupId() == 0)
+ .collect(Collectors.toSet()), Set.of(committingSegment, newConsumingSegment));
+ assertEquals(new HashSet<>(segmentManager._idealState.getInstanceStateMap(newConsumingSegment).values()),
+ Set.of(SegmentStateModel.CONSUMING));
+ }
+
+ @Test
+ public void testCommitStartSucceedsWhenFailedIdealStateUpdateWasApplied() {
+ 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();
+
+ segmentManager._numAppliedButFailedIdealStateUpdates = 1;
+ segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
+ createCommittingSegmentDescriptor(committingSegment));
+
+ assertEquals(segmentManager._segmentZKMetadataMap.get(committingSegment).getStatus(), Status.COMMITTING);
+ assertEquals(segmentManager._segmentZKMetadataMap.get(newConsumingSegment).getStatus(), Status.IN_PROGRESS);
+ 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(newConsumingSegment).values()),
+ Set.of(SegmentStateModel.CONSUMING));
+ ZkHelixPropertyStore<ZNRecord> propertyStore = segmentManager._mockResourceManager.getPropertyStore();
+ verify(propertyStore, never()).remove(anyString(), eq(AccessOption.PERSISTENT));
+ }
+
+ @Test
+ public void testCommitStartRetryAfterCompletedCommitStartDoesNotOverwriteNewSegment() {
+ 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.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME, committingSegmentDescriptor);
+ int newSegmentZKMetadataVersion = segmentManager._segmentZKMetadataVersionMap.get(newConsumingSegment);
+ Map<String, Map<String, String>> expectedInstanceStatesMap =
+ cloneInstanceStatesMap(segmentManager._idealState.getRecord().getMapFields());
+
+ // E.g. the server retries the commit after the commit end failed. The new segment might be consuming already.
+ segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME, committingSegmentDescriptor);
+ assertEquals((int) segmentManager._segmentZKMetadataVersionMap.get(newConsumingSegment),
+ newSegmentZKMetadataVersion);
+ assertEquals(segmentManager._idealState.getRecord().getMapFields(), expectedInstanceStatesMap);
+ }
+
+ @Test
+ public void testCommitStartDoesNotOverwriteNewSegmentCreatedByConcurrentAttempt() {
+ 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();
+
+ // After step 1, another controller (e.g. the new lead controller) resumes and completes the commit start, and the
+ // new segment is committed already
+ segmentManager._beforeNewSegmentZKMetadata =
+ () -> completeCommitStartElsewhere(segmentManager, committingSegment, newConsumingSegment, Status.DONE);
+ segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
+ createCommittingSegmentDescriptor(committingSegment));
+
+ assertEquals(segmentManager._segmentZKMetadataMap.get(newConsumingSegment).getStatus(), Status.DONE);
+ assertEquals(new HashSet<>(segmentManager._idealState.getInstanceStateMap(committingSegment).values()),
+ Set.of(SegmentStateModel.ONLINE));
+ }
+
+ @Test
+ public void testCommitStartFailureRemovesNewSegmentAfterPartitionRepairedWithAnotherSegment() {
+ 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 repairSegment =
+ new LLCSegmentName(RAW_TABLE_NAME, 0, 1, CURRENT_TIME_MS + TimeUnit.MINUTES.toMillis(2)).getSegmentName();
+
+ // Before the IdealState update, RealtimeSegmentValidationManager repairs the partition with another new segment,
+ // so the new segment of this attempt can no longer be added
+ segmentManager._beforeCommitIdealStateUpdate =
+ () -> completeCommitStartElsewhere(segmentManager, committingSegment, repairSegment, Status.IN_PROGRESS);
+ Map<String, Map<String, String>> instanceStatesMap = segmentManager._idealState.getRecord().getMapFields();
+ assertThrows(RuntimeException.class,
+ () -> segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
+ createCommittingSegmentDescriptor(committingSegment)));
+ assertFalse(segmentManager._segmentZKMetadataMap.containsKey(newConsumingSegment));
+ assertFalse(instanceStatesMap.containsKey(newConsumingSegment));
+ assertEquals(new HashSet<>(instanceStatesMap.get(repairSegment).values()), Set.of(SegmentStateModel.CONSUMING));
+ assertEquals(segmentManager._segmentZKMetadataMap.get(committingSegment).getStatus(), Status.COMMITTING);
+ }
+
+ @Test
+ public void testValidationAfterCommitStartAndConcurrentRepairCreatedDifferentNewSegments() {
+ 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 repairSegment =
+ new LLCSegmentName(RAW_TABLE_NAME, 0, 1, CURRENT_TIME_MS + TimeUnit.MINUTES.toMillis(30)).getSegmentName();
+
+ // The commit start keeps its new segment after its IdealState update fails, then a concurrent repair that did not
+ // see that segment adds another new segment with the same sequence number
+ segmentManager._numFailedIdealStateUpdates = 1;
+ assertThrows(RuntimeException.class,
+ () -> segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
+ createCommittingSegmentDescriptor(committingSegment)));
+ assertTrue(segmentManager._segmentZKMetadataMap.containsKey(newConsumingSegment));
+ completeCommitStartElsewhere(segmentManager, committingSegment, repairSegment, Status.IN_PROGRESS);
+ Map<String, Map<String, String>> expectedInstanceStatesMap =
+ cloneInstanceStatesMap(segmentManager._idealState.getRecord().getMapFields());
+
+ // Validation keeps the segment in the IdealState as the latest segment, whichever order the segments are listed in
+ segmentManager._exceededMaxSegmentCompletionTime = true;
+ for (String lastSegment : List.of(newConsumingSegment, repairSegment)) {
+ Map<String, SegmentZKMetadata> segmentZKMetadataMap = new LinkedHashMap<>(segmentManager._segmentZKMetadataMap);
+ segmentZKMetadataMap.put(lastSegment, segmentZKMetadataMap.remove(lastSegment));
+ segmentManager._segmentZKMetadataMap = segmentZKMetadataMap;
+ segmentManager.ensureAllPartitionsConsuming();
+ assertEquals(segmentManager._idealState.getRecord().getMapFields(), expectedInstanceStatesMap);
+ }
+ }
+
+ @Test
+ public void testCommitStartUsesNewSegmentCreatedConcurrentlyByAnotherAttempt() {
+ FakePinotLLCRealtimeSegmentManager segmentManager = setUpTableForCommitStart();
+ segmentManager._useProductionCreateIfAbsent = true;
+ 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 newConsumingSegmentPath =
+ ZKMetadataProvider.constructPropertyStorePathForSegment(REALTIME_TABLE_NAME, newConsumingSegment);
+ // Another attempt creates the new segment ZK metadata between the existence check and the creation
+ ZkHelixPropertyStore<ZNRecord> propertyStore = segmentManager._mockResourceManager.getPropertyStore();
+ when(propertyStore.create(eq(newConsumingSegmentPath), any(), eq(AccessOption.PERSISTENT))).thenReturn(false);
+ when(propertyStore.exists(newConsumingSegmentPath, AccessOption.PERSISTENT)).thenReturn(true);
+
+ segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
+ createCommittingSegmentDescriptor(committingSegment));
+ verify(propertyStore).create(eq(newConsumingSegmentPath), any(), eq(AccessOption.PERSISTENT));
+ verify(propertyStore, never()).set(eq(newConsumingSegmentPath), any(), anyInt(), anyInt());
+ assertFalse(segmentManager._segmentZKMetadataMap.containsKey(newConsumingSegment));
+ assertEquals(new HashSet<>(segmentManager._idealState.getInstanceStateMap(newConsumingSegment).values()),
+ Set.of(SegmentStateModel.CONSUMING));
+ }
+
+ @Test
+ public void testCreateSegmentZKMetadataIfAbsentFailsWhenNotCreated() {
+ FakePinotLLCRealtimeSegmentManager segmentManager = setUpTableForCommitStart();
+ segmentManager._useProductionCreateIfAbsent = true;
+ ZkHelixPropertyStore<ZNRecord> propertyStore = segmentManager._mockResourceManager.getPropertyStore();
+ when(propertyStore.create(anyString(), any(), eq(AccessOption.PERSISTENT))).thenReturn(false);
+ SegmentZKMetadata segmentZKMetadata =
+ new SegmentZKMetadata(new LLCSegmentName(RAW_TABLE_NAME, 0, 1, CURRENT_TIME_MS).getSegmentName());
+ assertThrows(IllegalStateException.class,
+ () -> segmentManager.createSegmentZKMetadataIfAbsent(REALTIME_TABLE_NAME, segmentZKMetadata));
+ }
+
+ @Test
+ public void testCommitStartDoesNotResumeWithDifferentEndOffset() {
+ FakePinotLLCRealtimeSegmentManager segmentManager = setUpTableForCommitStart();
+ String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
+ segmentManager._numFailedIdealStateUpdates = 1;
+ assertThrows(RuntimeException.class,
+ () -> segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
+ createCommittingSegmentDescriptor(committingSegment)));
+
+ String otherOffset = new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS + 1L).toString();
+ IllegalStateException e = expectThrows(IllegalStateException.class,
+ () -> segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
+ createCommittingSegmentDescriptor(committingSegment, otherOffset)));
+ assertTrue(e.getMessage().contains("should be IN_PROGRESS, found: COMMITTING"));
+ }
+
+ @Test
+ public void testCommitStartDoesNotResumeAfterMaxSegmentCompletionTime() {
+ 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));
+
+ // RealtimeSegmentValidationManager owns the repair once the max segment completion time is exceeded
+ segmentManager._exceededMaxSegmentCompletionTime = true;
+ assertThrows(IllegalStateException.class,
+ () -> segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME, committingSegmentDescriptor));
+ assertFalse(segmentManager._idealState.getRecord().getMapFields().containsKey(newConsumingSegment));
+ assertEquals(segmentManager._segmentZKMetadataMap.get(committingSegment).getStatus(), Status.COMMITTING);
+ }
+
+ /// 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
+ @SuppressWarnings("unchecked")
+ public void testIdealStateUpdateOnSegmentCompletionSkipsAppliedUpdate() {
+ for (boolean groupCommitEnabled : new boolean[]{true, false}) {
+ ControllerConf controllerConf = new ControllerConf();
+ controllerConf.setDataDir(TEMP_DIR.toString());
+ controllerConf.setProperty(ControllerConf.CONTROLLER_SEGMENT_COMPLETION_GROUP_COMMIT_ENABLED,
+ groupCommitEnabled);
+ PinotHelixResourceManager resourceManager = FakePinotLLCRealtimeSegmentManager.createMockedResourceManager();
+ when(resourceManager.getPropertyStore().create(anyString(), any(), eq(AccessOption.PERSISTENT))).thenReturn(
+ true);
+ FakePinotLLCRealtimeSegmentManager segmentManager =
+ new FakePinotLLCRealtimeSegmentManager(resourceManager, controllerConf);
+ setUpNewTable(segmentManager, 2, 5, 4);
+ segmentManager._useProductionIdealStateUpdate = true;
+
+ HelixDataAccessor dataAccessor = mock(HelixDataAccessor.class);
+ when(resourceManager.getHelixZkManager().getHelixDataAccessor()).thenReturn(dataAccessor);
+ when(dataAccessor.keyBuilder()).thenReturn(new PropertyKey.Builder(CLUSTER_NAME));
+ int[] version = {0};
+ when(dataAccessor.getProperty(any(PropertyKey.class))).thenAnswer(invocation -> {
+ IdealState idealState = new IdealState(segmentManager._idealState.getRecord());
+ idealState.getRecord().setVersion(version[0]);
+ return idealState;
+ });
+ BaseDataAccessor<ZNRecord> baseDataAccessor = mock(BaseDataAccessor.class);
+ when(dataAccessor.getBaseDataAccessor()).thenReturn(baseDataAccessor);
+ when(baseDataAccessor.set(anyString(), any(ZNRecord.class), anyInt(), eq(AccessOption.PERSISTENT))).thenAnswer(
+ invocation -> {
+ assertEquals((int) invocation.getArgument(2), version[0]);
+ segmentManager._idealState = new IdealState((ZNRecord) invocation.getArgument(1));
+ version[0]++;
+ // The write is applied, but the response is lost
+ return false;
+ });
+
+ 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();
+ segmentManager.commitSegmentMetadataToCommitting(REALTIME_TABLE_NAME,
+ createCommittingSegmentDescriptor(committingSegment));
+
+ // The retry finds the update applied and does not write again
+ verify(baseDataAccessor, times(1)).set(anyString(), any(ZNRecord.class), anyInt(), eq(AccessOption.PERSISTENT));
+ verify(dataAccessor, times(2)).getProperty(any(PropertyKey.class));
+ 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(newConsumingSegment).values()),
+ Set.of(SegmentStateModel.CONSUMING));
+ }
+ }
+
+ @Test
+ public void testIsSegmentCompletionApplied() {
+ String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
+ String newSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 1, CURRENT_TIME_MS).getSegmentName();
+ IdealState idealState = new IdealState(REALTIME_TABLE_NAME);
+
+ // Committing segment missing
+ assertFalse(PinotLLCRealtimeSegmentManager.isSegmentCompletionApplied(idealState, committingSegment, newSegment));
+
+ // Committing segment still CONSUMING on one replica
+ idealState.setPartitionState(committingSegment, "Server_0", SegmentStateModel.ONLINE);
+ idealState.setPartitionState(committingSegment, "Server_1", SegmentStateModel.CONSUMING);
+ assertFalse(PinotLLCRealtimeSegmentManager.isSegmentCompletionApplied(idealState, committingSegment, newSegment));
+
+ // Committing segment ONLINE, new segment missing
+ idealState.setPartitionState(committingSegment, "Server_1", SegmentStateModel.ONLINE);
+ assertFalse(PinotLLCRealtimeSegmentManager.isSegmentCompletionApplied(idealState, committingSegment, newSegment));
+ assertTrue(PinotLLCRealtimeSegmentManager.isSegmentCompletionApplied(idealState, committingSegment, null));
+
+ // Committing segment ONLINE, new segment present
+ idealState.setPartitionState(newSegment, "Server_0", SegmentStateModel.CONSUMING);
+ assertTrue(PinotLLCRealtimeSegmentManager.isSegmentCompletionApplied(idealState, committingSegment, newSegment));
+
+ // Table paused, so no new segment is expected
+ idealState.getRecord().getMapFields().remove(newSegment);
+ idealState.getRecord().setSimpleField(PinotLLCRealtimeSegmentManager.PAUSE_STATE,
+ new PauseState(true, PauseState.ReasonCode.ADMINISTRATIVE, "pause-for-test", Long.toString(CURRENT_TIME_MS),
+ List.of()).toJsonString());
+ assertTrue(PinotLLCRealtimeSegmentManager.isSegmentCompletionApplied(idealState, committingSegment, newSegment));
+ }
+
/// Test cases for fixing LLC segment by uploading to segment store if missing
@Test
public void testUploadToSegmentStore()
@@ -2436,10 +2837,26 @@ public class PinotLLCRealtimeSegmentManagerTest {
InstancePartitions _consumingInstancePartitions;
Map<String, SegmentZKMetadata> _segmentZKMetadataMap = new HashMap<>();
Map<String, Integer> _segmentZKMetadataVersionMap = new HashMap<>();
+ Map<String, Long> _segmentZKMetadataModifiedTimeMap = new HashMap<>();
+ long _currentTimeMs = CURRENT_TIME_MS;
IdealState _idealState;
int _numPartitions;
List<StreamMetadata> _streamMetadataList = null;
boolean _exceededMaxSegmentCompletionTime = 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
+ int _numAppliedButFailedIdealStateUpdates = 0;
+ // Whether to update the IdealState on segment completion through the production code path
+ boolean _useProductionIdealStateUpdate = false;
+ // Whether to create segment ZK metadata if absent through the production code path
+ boolean _useProductionCreateIfAbsent = false;
+ // Runs before creating the new segment ZK metadata (step 2) of a segment commit
+ Runnable _beforeNewSegmentZKMetadata = () -> {
+ };
+ // Runs before updating the IdealState (step 3) of a segment commit
+ Runnable _beforeCommitIdealStateUpdate = () -> {
+ };
FileUploadDownloadClient _mockedFileUploadDownloadClient;
PinotHelixResourceManager _mockResourceManager;
@@ -2556,6 +2973,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
Preconditions.checkState(_segmentZKMetadataMap.containsKey(segmentName));
if (stat != null) {
stat.setVersion(_segmentZKMetadataVersionMap.get(segmentName));
+ stat.setMtime(_segmentZKMetadataModifiedTimeMap.getOrDefault(segmentName, CURRENT_TIME_MS));
}
return new SegmentZKMetadata(new ZNRecord(_segmentZKMetadataMap.get(segmentName).toZNRecord()));
}
@@ -2569,6 +2987,34 @@ public class PinotLLCRealtimeSegmentManagerTest {
}
_segmentZKMetadataMap.put(segmentName, segmentZKMetadata);
_segmentZKMetadataVersionMap.put(segmentName, version + 1);
+ _segmentZKMetadataModifiedTimeMap.put(segmentName, _currentTimeMs);
+ }
+
+ @Override
+ boolean segmentZKMetadataExists(String realtimeTableName, String segmentName) {
+ return _segmentZKMetadataMap.containsKey(segmentName);
+ }
+
+ @Override
+ boolean createSegmentZKMetadataIfAbsent(String realtimeTableName, SegmentZKMetadata segmentZKMetadata) {
+ if (_useProductionCreateIfAbsent) {
+ return super.createSegmentZKMetadataIfAbsent(realtimeTableName, segmentZKMetadata);
+ }
+ if (_segmentZKMetadataMap.containsKey(segmentZKMetadata.getSegmentName())) {
+ return false;
+ }
+ persistSegmentZKMetadata(realtimeTableName, segmentZKMetadata, -1);
+ return true;
+ }
+
+ @Override
+ protected void preProcessNewSegmentZKMetadata() {
+ _beforeNewSegmentZKMetadata.run();
+ }
+
+ @Override
+ protected void preProcessCommitIdealStateUpdate() {
+ _beforeCommitIdealStateUpdate.run();
}
@Override
@@ -2585,6 +3031,14 @@ public class PinotLLCRealtimeSegmentManagerTest {
IdealState updateIdealStateOnSegmentCompletion(String realtimeTableName, String committingSegmentName,
String newSegmentName, SegmentAssignment segmentAssignment,
Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap) {
+ if (_useProductionIdealStateUpdate) {
+ return super.updateIdealStateOnSegmentCompletion(realtimeTableName, committingSegmentName, newSegmentName,
+ segmentAssignment, instancePartitionsMap);
+ }
+ if (_numFailedIdealStateUpdates > 0) {
+ _numFailedIdealStateUpdates--;
+ throw new RuntimeException("Injected IdealState update failure");
+ }
Map<String, String> committingSegmentInstanceStateMap = _idealState.getInstanceStateMap(committingSegmentName);
Preconditions.checkState(
committingSegmentInstanceStateMap != null && committingSegmentInstanceStateMap.containsValue(
@@ -2593,6 +3047,10 @@ public class PinotLLCRealtimeSegmentManagerTest {
updateInstanceStatesForNewConsumingSegment(_idealState.getRecord().getMapFields(), committingSegmentName,
isTablePaused(_idealState) || isTopicPaused(_idealState, committingSegmentName) ? null : newSegmentName,
segmentAssignment, instancePartitionsMap);
+ if (_numAppliedButFailedIdealStateUpdates > 0) {
+ _numAppliedButFailedIdealStateUpdates--;
+ throw new RuntimeException("Injected IdealState update failure after applying the update");
+ }
return _idealState;
}
@@ -2633,7 +3091,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
@Override
long getCurrentTimeMs() {
- return CURRENT_TIME_MS;
+ return _currentTimeMs;
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org