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 <[email protected]>
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: [email protected]
For additional commands, e-mail: [email protected]
