xiangfu0 commented on code in PR #19901:
URL: https://github.com/apache/pinot/pull/19901#discussion_r4291429713
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManager.java:
##########
@@ -383,6 +408,17 @@ private void computeTablePartitionReplicatedServersInfo() {
//noinspection SlowListContainsAll
if
(onlineServers.containsAll(partitionInfo._fullyReplicatedServers)) {
partitionInfo._segments.add(segment);
+ } else if (segmentInfo.hasAllAssignedReplicasOnline()) {
Review Comment:
Nice, this is the heart of the fix and reads clearly. One thing to confirm:
when this branch runs, `retainAll` can empty `fullyReplicatedServers`, and the
downstream `WorkerManager` then throws "Failed to find enabled fully replicated
server" for a partition-hinted leaf. That is the intended trade (a loud error
instead of silently dropped rows), and the PR body documents it well — just
flagging that operators of partitioned tables without partition-aware placement
will start seeing that error immediately rather than after the new-segment
window.
##########
pinot-broker/src/test/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManagerTest.java:
##########
@@ -295,6 +295,61 @@ public void
testPartitionMetadataManagerProcessingThroughSegmentChangesSinglePar
assertEquals(tablePartitionReplicatedServersInfo.getSegmentsWithInvalidPartition().get(0),
segmentInvalid);
}
+ /// A new segment is deferred only while its replicas are still coming up.
One that is already online on every server
+ /// the ideal state assigns, but on other servers than the rest of its
partition, will never cover them: deferring it
+ /// made a partition-hinted query silently skip its rows until it stopped
being new. It must instead count like an
+ /// old segment, so the partition reports that no server holds all of it.
+ @Test
+ public void testFullyOnlineNewSegmentOnOtherServersIsNotDeferred() {
+ ExternalView externalView = new ExternalView(OFFLINE_TABLE_NAME);
+ Map<String, Map<String, String>> segmentAssignment =
externalView.getRecord().getMapFields();
+ IdealState idealState = new IdealState(OFFLINE_TABLE_NAME);
+ Map<String, Map<String, String>> idealAssignment =
idealState.getRecord().getMapFields();
+ Set<String> onlineSegments = new HashSet<>();
+
+ SegmentPartitionMetadataManager partitionMetadataManager =
+ new SegmentPartitionMetadataManager(OFFLINE_TABLE_NAME,
PARTITION_COLUMN, PARTITION_COLUMN_FUNC, NUM_PARTITIONS,
+ TimeUnit.MINUTES.toMillis(5));
+ SegmentZkMetadataFetcher segmentZkMetadataFetcher =
+ new SegmentZkMetadataFetcher(OFFLINE_TABLE_NAME, _propertyStore);
+ segmentZkMetadataFetcher.register(partitionMetadataManager);
+ segmentZkMetadataFetcher.init(idealState, externalView, onlineSegments);
+
+ // An old segment of partition 0 on server0.
+ String oldSegment = "oldSegment";
+ onlineSegments.add(oldSegment);
+ idealAssignment.put(oldSegment, Map.of(SERVER_0, ONLINE));
+ segmentAssignment.put(oldSegment, Map.of(SERVER_0, ONLINE));
+ setSegmentZKMetadata(oldSegment, PARTITION_COLUMN_FUNC, NUM_PARTITIONS, 0,
0L);
+
+ // A new segment of partition 0 still coming up: assigned to server0 and
server1, online on server1 only.
Review Comment:
This test nicely covers the two main cases. Would you consider adding one
more for the IS/EV divergence window — a segment whose IS assigns {server0,
server1} but whose EV shows server1 plus a stale server0 entry while the real
new replica is still down? That is the case where the count comparison in
`hasAllAssignedReplicasOnline()` can admit a segment that is still coming up,
so a test there would pin down the intended behaviour.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManager.java:
##########
@@ -230,12 +247,20 @@ public synchronized void onAssignmentChange(IdealState
idealState, ExternalView
LOGGER.error("Failed to find segment info for segment: {} in table: {}
while handling assignment change",
segment, _tableNameWithType);
segmentInfo = new SegmentInfo(INVALID_PARTITION_ID,
INVALID_CREATION_TIME_MS,
- getOnlineServers(instanceStateMap, onlineServersCache));
+ getOnlineServers(instanceStateMap, onlineServersCache),
+ getNumAssignedReplicas(idealState.getInstanceStateMap(segment)));
_segmentInfoMap.put(segment, segmentInfo);
partitionsChanged = true;
- } else if (!hasSameOnlineServers(segmentInfo._onlineServers,
instanceStateMap)) {
- segmentInfo._onlineServers = getOnlineServers(instanceStateMap,
onlineServersCache);
- onlineServersChanged = true;
+ } else {
+ if (!hasSameOnlineServers(segmentInfo._onlineServers,
instanceStateMap)) {
+ segmentInfo._onlineServers = getOnlineServers(instanceStateMap,
onlineServersCache);
+ onlineServersChanged = true;
+ }
+ int numAssignedReplicas =
getNumAssignedReplicas(idealState.getInstanceStateMap(segment));
+ if (segmentInfo._numAssignedReplicas != numAssignedReplicas) {
+ segmentInfo._numAssignedReplicas = numAssignedReplicas;
+ onlineServersChanged = true;
Review Comment:
Setting `onlineServersChanged = true` here correctly forces a recompute when
only the assigned-replica count changed, so the behaviour is right. The flag
name is now a little misleading though, since it no longer strictly means the
online servers changed — maybe rename to something like `recomputeNeeded` or
add a short comment so it doesn't trip someone up later.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManager.java:
##########
@@ -458,11 +494,20 @@ private static class SegmentInfo {
int _partitionId;
long _creationTimeMs;
List<String> _onlineServers;
+ /// Replicas the ideal state assigns in ONLINE or CONSUMING state; 0 when
the ideal state does not list the segment.
+ int _numAssignedReplicas;
- SegmentInfo(int partitionId, long creationTimeMs, List<String>
onlineServers) {
+ SegmentInfo(int partitionId, long creationTimeMs, List<String>
onlineServers, int numAssignedReplicas) {
_partitionId = partitionId;
_creationTimeMs = creationTimeMs;
_onlineServers = onlineServers;
+ _numAssignedReplicas = numAssignedReplicas;
+ }
+
+ /// Whether the segment is online on as many servers as the ideal state
assigns it to, i.e. it is not still coming
+ /// up. False when the ideal state does not list it, which keeps treating
such a segment as not ready.
+ boolean hasAllAssignedReplicasOnline() {
+ return _numAssignedReplicas > 0 && _onlineServers.size() >=
_numAssignedReplicas;
Review Comment:
This readiness check compares an ExternalView count
(`_onlineServers.size()`) against an IdealState count (`_numAssignedReplicas`),
and the two come from eventually-consistent sources with overlapping but not
identical server sets. During a reassignment or drain, the EV can briefly still
list an old ONLINE server that the IS no longer assigns, so the size can reach
the assigned count while a genuinely assigned replica is still coming up —
which is exactly the case this PR wants to keep deferring.
Could we test set-containment instead, i.e. that every IS-assigned server is
present in `_onlineServers`? That would need keeping the assigned server set
rather than just the count, but it removes the cross-source mismatch. The `>=`
itself looks like the right choice given a count compare (it tolerates an extra
EV server); the concern is the count compare, not the operator.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManager.java:
##########
@@ -173,6 +174,21 @@ private static boolean isOnline(String instanceState) {
return instanceState.equals(SegmentStateModel.ONLINE) ||
instanceState.equals(SegmentStateModel.CONSUMING);
}
+ /// Returns the number of servers the ideal state assigns the segment to in
ONLINE or CONSUMING state, or 0 when the
+ /// ideal state does not list it.
+ private static int getNumAssignedReplicas(@Nullable Map<String, String>
idealInstanceStateMap) {
+ if (idealInstanceStateMap == null) {
+ return 0;
+ }
+ int numAssignedReplicas = 0;
+ for (String instanceState : idealInstanceStateMap.values()) {
+ if (isOnline(instanceState)) {
+ numAssignedReplicas++;
+ }
+ }
+ return numAssignedReplicas;
Review Comment:
Small thing worth a comment: the `isOnline` filter here means a segment
whose IS replicas are all OFFLINE (e.g. being dropped) gets
`_numAssignedReplicas == 0`, so `hasAllAssignedReplicasOnline()` returns false
and it stays deferred. That looks like the safe fallback you intend, but it's
subtle enough that a one-line note on why 0 means "keep deferring" would help
the next reader.
--
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]
