xiangfu0 commented on code in PR #19916:
URL: https://github.com/apache/pinot/pull/19916#discussion_r4332397690
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeSegmentAssignment.java:
##########
@@ -158,8 +178,100 @@ protected List<String> assignConsumingSegment(int
segmentPartitionId, InstancePa
}
}
+ /// Assigns a CONSUMING segment of a partition-pinned table to the instances
of its partition, see
+ /// [BaseSegmentAssignment#getOrCreatePartitionInstances].
+ private List<String> assignPinnedConsumingSegment(String segmentName,
IdealState idealState,
+ InstancePartitions instancePartitions) {
+ Integer partitionId =
+ SegmentUtils.getSegmentPartitionId(segmentName, _tableNameWithType,
_helixManager, _partitionColumn);
+ Preconditions.checkState(partitionId != null, "Failed to find partition id
for segment: %s of table: %s",
+ segmentName, _tableNameWithType);
+ return getOrCreatePartitionInstances(idealState, partitionId,
Review Comment:
**Once a partition has an entry, the placement computed from the instance
partitions is discarded, so evacuating servers via the instance partitions no
longer redirects new segments.** Example: non-upsert strictReplicaGroup
realtime table (newly pinned by this PR); partition p3's servers fail and mark
the consuming replicas OFFLINE; the operator swaps them with `replaceInstance`
without a full rebalance; ensureAllPartitionsConsuming ->
createNewConsumingSegment -> assignSegment returns the entry, so the new
CONSUMING segment goes back to the dead servers and p3 stays down until a
rebalance (base: placed on the replacement servers). Uploads to qualifying
offline tables behave the same. Consider letting the candidate win when the
entry's instances are no longer in the instance partitions.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -1622,174 +1656,280 @@ private static void handleErrorInstance(String
segmentName, String instanceName,
}
}
+ /// Returns the target assignment for the ideal state changed while waiting
for the external view to converge, when
+ /// the segments to be moved are unchanged: the current assignment with the
target of the segments to be moved. For
+ /// a partition-pinned table (strict replica-group routing), a partition
added while waiting keeps its current
+ /// instances, and a segment added while waiting, which is assigned to the
current instances of its partition, is
+ /// moved to the target instances of its partition, unless it is a CONSUMING
segment and CONSUMING segments are not
+ /// included in the rebalance. Returns `null` when the target assignment
must be recalculated instead: when the
+ /// partition id of such a segment is not in its name, or when CONSUMING
segments are not included and an ONLINE
+ /// segment was added to a pauseless table, because it can still be
committing (which only its ZK metadata tells),
+ /// in which case it is not moved either.
+ @Nullable
+ @VisibleForTesting
+ static TableAssignment getTargetAssignmentForNewIdealState(Map<String,
Map<String, String>> currentAssignment,
+ Map<String, Map<String, String>> targetAssignment, Map<Integer,
List<String>> currentPartitionAssignment,
+ Map<Integer, List<String>> targetPartitionAssignment, List<String>
segmentsToMove,
+ boolean enableStrictReplicaGroup, boolean includeConsuming, boolean
pauseless, Logger tableRebalanceLogger) {
+ // Other instance assignment code returns a TreeMap to keep it sorted,
doing the same here
+ Map<String, Map<String, String>> newTargetAssignment = new
TreeMap<>(currentAssignment);
+ for (String segment : segmentsToMove) {
+ newTargetAssignment.put(segment, targetAssignment.get(segment));
+ }
+ if (!enableStrictReplicaGroup) {
+ return new TableAssignment(newTargetAssignment,
targetPartitionAssignment);
+ }
+ Map<Integer, List<String>> newTargetPartitionAssignment = new
TreeMap<>(targetPartitionAssignment);
+ for (Map.Entry<Integer, List<String>> entry :
currentPartitionAssignment.entrySet()) {
+ newTargetPartitionAssignment.putIfAbsent(entry.getKey(),
entry.getValue());
+ }
+ for (Map.Entry<String, Map<String, String>> entry :
currentAssignment.entrySet()) {
+ String segment = entry.getKey();
+ if (targetAssignment.containsKey(segment)) {
+ continue;
+ }
+ Map<String, String> currentInstanceStateMap = entry.getValue();
+ String state;
+ if (currentInstanceStateMap.containsValue(SegmentStateModel.ONLINE)) {
+ if (!includeConsuming && pauseless) {
+ tableRebalanceLogger.info("Found new ONLINE segment: {} of pauseless
table while CONSUMING segments are not "
+ + "included, re-calculating the target assignment based on the
new IdealState", segment);
+ return null;
+ }
+ state = SegmentStateModel.ONLINE;
+ } else if
(currentInstanceStateMap.containsValue(SegmentStateModel.CONSUMING)) {
+ if (!includeConsuming) {
+ continue;
+ }
+ state = SegmentStateModel.CONSUMING;
+ } else {
+ // All instances are OFFLINE, and OFFLINE segments are not rebalanced
+ continue;
+ }
+ Integer partitionId =
SegmentUtils.getPartitionIdFromSegmentName(segment);
+ if (partitionId == null) {
Review Comment:
**Every OFFLINE upload during a strictReplicaGroup rebalance forces a full
recompute, which also misplaces tier-eligible uploads.** This check is gated on
routing (`enableStrictReplicaGroup`), not on the table being pinned, and
offline names never carry a partition id, so it returns null for each upload;
the base took the copy path for offline tables. (1) The recompute reruns
getInstancePartitionsMap (persisted with reassignInstances) and rebalanceTable;
for a non-pinned replica-group table (numPartitions>1) the strategy does one
uncached ZK read per segment, and with continuous pushes the versioned write
keeps losing (ZkBadVersion only rolls back and loops), so the rebalance can
stall. (2) The recompute reuses the tier cache frozen at rebalance start
(`_segmentToTierName`; TierSegmentAssignment treats absent segments as
non-tier), so a tier-eligible upload that tiered segment assignment placed on
tier servers is retargeted to the default instances, then moved back by the
next Seg
mentRelocator run.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -1622,174 +1656,280 @@ private static void handleErrorInstance(String
segmentName, String instanceName,
}
}
+ /// Returns the target assignment for the ideal state changed while waiting
for the external view to converge, when
+ /// the segments to be moved are unchanged: the current assignment with the
target of the segments to be moved. For
+ /// a partition-pinned table (strict replica-group routing), a partition
added while waiting keeps its current
+ /// instances, and a segment added while waiting, which is assigned to the
current instances of its partition, is
+ /// moved to the target instances of its partition, unless it is a CONSUMING
segment and CONSUMING segments are not
+ /// included in the rebalance. Returns `null` when the target assignment
must be recalculated instead: when the
+ /// partition id of such a segment is not in its name, or when CONSUMING
segments are not included and an ONLINE
+ /// segment was added to a pauseless table, because it can still be
committing (which only its ZK metadata tells),
+ /// in which case it is not moved either.
+ @Nullable
+ @VisibleForTesting
+ static TableAssignment getTargetAssignmentForNewIdealState(Map<String,
Map<String, String>> currentAssignment,
+ Map<String, Map<String, String>> targetAssignment, Map<Integer,
List<String>> currentPartitionAssignment,
+ Map<Integer, List<String>> targetPartitionAssignment, List<String>
segmentsToMove,
+ boolean enableStrictReplicaGroup, boolean includeConsuming, boolean
pauseless, Logger tableRebalanceLogger) {
+ // Other instance assignment code returns a TreeMap to keep it sorted,
doing the same here
+ Map<String, Map<String, String>> newTargetAssignment = new
TreeMap<>(currentAssignment);
+ for (String segment : segmentsToMove) {
+ newTargetAssignment.put(segment, targetAssignment.get(segment));
+ }
+ if (!enableStrictReplicaGroup) {
+ return new TableAssignment(newTargetAssignment,
targetPartitionAssignment);
+ }
+ Map<Integer, List<String>> newTargetPartitionAssignment = new
TreeMap<>(targetPartitionAssignment);
+ for (Map.Entry<Integer, List<String>> entry :
currentPartitionAssignment.entrySet()) {
+ newTargetPartitionAssignment.putIfAbsent(entry.getKey(),
entry.getValue());
+ }
+ for (Map.Entry<String, Map<String, String>> entry :
currentAssignment.entrySet()) {
+ String segment = entry.getKey();
+ if (targetAssignment.containsKey(segment)) {
+ continue;
+ }
+ Map<String, String> currentInstanceStateMap = entry.getValue();
+ String state;
+ if (currentInstanceStateMap.containsValue(SegmentStateModel.ONLINE)) {
+ if (!includeConsuming && pauseless) {
+ tableRebalanceLogger.info("Found new ONLINE segment: {} of pauseless
table while CONSUMING segments are not "
+ + "included, re-calculating the target assignment based on the
new IdealState", segment);
+ return null;
+ }
+ state = SegmentStateModel.ONLINE;
+ } else if
(currentInstanceStateMap.containsValue(SegmentStateModel.CONSUMING)) {
+ if (!includeConsuming) {
+ continue;
+ }
+ state = SegmentStateModel.CONSUMING;
+ } else {
+ // All instances are OFFLINE, and OFFLINE segments are not rebalanced
+ continue;
+ }
+ Integer partitionId =
SegmentUtils.getPartitionIdFromSegmentName(segment);
+ if (partitionId == null) {
+ tableRebalanceLogger.info("Found new segment: {} without partition id
in its name, re-calculating the target "
+ + "assignment based on the new IdealState", segment);
+ return null;
+ }
+ List<String> currentInstances =
currentPartitionAssignment.get(partitionId);
+ if (currentInstances == null || currentInstances.size() !=
currentInstanceStateMap.size()
Review Comment:
**A segment added mid-rebalance to a partition without an entry is left in
place, whereas the base always fully recomputed for strict (upsert/dedup)
tables.** Example (rolling upgrade): pre-PR upsert table with no entries; a new
controller rebalances p0 [A,B]->[C,D]; the old-version table lead
resumes/repairs p0's consumption or accepts an upload for p0 and places the new
segment on [A,B] without creating an entry; no segment in segmentsToMove
changes state, so this branch keeps it at [A,B], the rest of p0 moves to [C,D],
the final step sets entry p0=[C,D], and the rebalance reports DONE with the
upsert partition split.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeSegmentAssignment.java:
##########
@@ -158,8 +178,100 @@ protected List<String> assignConsumingSegment(int
segmentPartitionId, InstancePa
}
}
+ /// Assigns a CONSUMING segment of a partition-pinned table to the instances
of its partition, see
+ /// [BaseSegmentAssignment#getOrCreatePartitionInstances].
+ private List<String> assignPinnedConsumingSegment(String segmentName,
IdealState idealState,
+ InstancePartitions instancePartitions) {
+ Integer partitionId =
+ SegmentUtils.getSegmentPartitionId(segmentName, _tableNameWithType,
_helixManager, _partitionColumn);
+ Preconditions.checkState(partitionId != null, "Failed to find partition id
for segment: %s of table: %s",
+ segmentName, _tableNameWithType);
+ return getOrCreatePartitionInstances(idealState, partitionId,
+ assignConsumingSegment(partitionId, instancePartitions));
+ }
+
+ /// OFFLINE segments are skipped because they are not rebalanced, so their
assignment can be stale. The instances are
+ /// taken from, in order:
+ /// - The CONSUMING segment of the partition, which is always hosted by the
CONSUMING instances of the partition.
+ /// - The latest LLC segment of the partition, else an uploaded segment of
the partition, whose partition id is in
+ /// its name or in its ZK metadata. Completed segments are not used when
they are relocated to the COMPLETED
+ /// instances, or when the segment has been moved to a tier, because they
are not on the CONSUMING instances of
+ /// the partition then. The tier is read from the segment ZK metadata
whether or not the table still has tiers
+ /// configured, because removing the tier config does not move the
segments back.
+ @Nullable
@Override
- public Map<String, Map<String, String>> rebalanceTable(Map<String,
Map<String, String>> currentAssignment,
+ protected Set<String> getExistingAssignment(int partitionId, Map<String,
Map<String, String>> currentAssignment) {
+ LLCSegmentName latestLLCSegmentName = null;
+ Map<String, String> latestInstanceStateMap = null;
+ List<String> uploadedSegments = new ArrayList<>();
+ for (Map.Entry<String, Map<String, String>> entry :
currentAssignment.entrySet()) {
+ String segmentName = entry.getKey();
+ Map<String, String> instanceStateMap = entry.getValue();
+ boolean completed;
+ if (instanceStateMap.containsValue(SegmentStateModel.ONLINE)) {
+ completed = true;
+ } else if (instanceStateMap.containsValue(SegmentStateModel.CONSUMING)) {
+ completed = false;
+ } else {
+ // All instances are OFFLINE, and OFFLINE segments are not rebalanced,
so their assignment can be stale
+ continue;
+ }
+ LLCSegmentName llcSegmentName = LLCSegmentName.of(segmentName);
+ if (llcSegmentName == null) {
+ uploadedSegments.add(segmentName);
+ continue;
+ }
+ if (llcSegmentName.getPartitionGroupId() != partitionId) {
+ continue;
+ }
+ if (!completed) {
+ return instanceStateMap.keySet();
+ }
+ if (latestLLCSegmentName == null
+ || llcSegmentName.getSequenceNumber() >
latestLLCSegmentName.getSequenceNumber()) {
+ latestLLCSegmentName = llcSegmentName;
+ latestInstanceStateMap = instanceStateMap;
+ }
+ }
+ // Completed segments relocated to the COMPLETED instances are not on the
CONSUMING instances of the partition
+ if
(InstanceAssignmentConfigUtils.shouldRelocateCompletedSegments(_tableConfig)) {
+ return null;
+ }
+ String latestSegmentName = null;
+ if (latestLLCSegmentName != null) {
+ latestSegmentName = latestLLCSegmentName.getSegmentName();
+ } else {
+ for (String uploadedSegment : uploadedSegments) {
+ Integer uploadedSegmentPartitionId =
+ SegmentUtils.getSegmentPartitionId(uploadedSegment,
_tableNameWithType, _helixManager, _partitionColumn);
Review Comment:
**When a partition has no entry and no ONLINE/CONSUMING LLC segment, this
reads the ZK metadata of every custom-named segment one by one inside the
IdealState updater, now for dedup and plain strictReplicaGroup tables too**
(base: only upsert scanned; dedup skipped uploaded segments; plain tables never
scanned). Example: plain/dedup strictReplicaGroup table with N MergeRollup
`merged_*` segments gains K stream partitions; the next
ensureAllPartitionsConsuming updater does K x N sequential ZK reads per attempt
(up to 10 attempts) while holding the table's group-commit queue, delaying
segment-commit updates; an orphan segment in the IdealState without ZK metadata
makes `SegmentUtils.getSegmentPartitionId` throw and the new partitions never
start consuming. Checking relocation first and batch-reading only the needed
segments outside the updater would avoid it.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -1622,174 +1656,280 @@ private static void handleErrorInstance(String
segmentName, String instanceName,
}
}
+ /// Returns the target assignment for the ideal state changed while waiting
for the external view to converge, when
+ /// the segments to be moved are unchanged: the current assignment with the
target of the segments to be moved. For
+ /// a partition-pinned table (strict replica-group routing), a partition
added while waiting keeps its current
+ /// instances, and a segment added while waiting, which is assigned to the
current instances of its partition, is
+ /// moved to the target instances of its partition, unless it is a CONSUMING
segment and CONSUMING segments are not
+ /// included in the rebalance. Returns `null` when the target assignment
must be recalculated instead: when the
+ /// partition id of such a segment is not in its name, or when CONSUMING
segments are not included and an ONLINE
+ /// segment was added to a pauseless table, because it can still be
committing (which only its ZK metadata tells),
+ /// in which case it is not moved either.
+ @Nullable
+ @VisibleForTesting
+ static TableAssignment getTargetAssignmentForNewIdealState(Map<String,
Map<String, String>> currentAssignment,
+ Map<String, Map<String, String>> targetAssignment, Map<Integer,
List<String>> currentPartitionAssignment,
+ Map<Integer, List<String>> targetPartitionAssignment, List<String>
segmentsToMove,
+ boolean enableStrictReplicaGroup, boolean includeConsuming, boolean
pauseless, Logger tableRebalanceLogger) {
+ // Other instance assignment code returns a TreeMap to keep it sorted,
doing the same here
+ Map<String, Map<String, String>> newTargetAssignment = new
TreeMap<>(currentAssignment);
+ for (String segment : segmentsToMove) {
+ newTargetAssignment.put(segment, targetAssignment.get(segment));
+ }
+ if (!enableStrictReplicaGroup) {
+ return new TableAssignment(newTargetAssignment,
targetPartitionAssignment);
+ }
+ Map<Integer, List<String>> newTargetPartitionAssignment = new
TreeMap<>(targetPartitionAssignment);
+ for (Map.Entry<Integer, List<String>> entry :
currentPartitionAssignment.entrySet()) {
+ newTargetPartitionAssignment.putIfAbsent(entry.getKey(),
entry.getValue());
Review Comment:
**putIfAbsent cannot tell a partition added while waiting from one the
target deliberately drops, so leftover entries can survive the rebalance.**
Example: formerly pinned OFFLINE strictReplicaGroup table whose instance
partitions now hold 2 instances per partition (rebalanceTable returns
`Map.of()`), entries {0:[A],1:[B]}; a retention deletion of a non-moving
segment lands before the first write, this makes target == current, and the
rebalance ends DONE with the stale entries; if the instance partitions are
later set back to one instance per partition via assignInstances/PUT without a
rebalance, uploads for partition 0 go to stale server A. Only re-adding
partitions that are new since the target was computed would avoid this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
