Jackie-Jiang commented on code in PR #19783:
URL: https://github.com/apache/pinot/pull/19783#discussion_r4253115082
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/timeboundary/TimeBoundaryManager.java:
##########
@@ -206,20 +208,59 @@ public synchronized void onAssignmentChange(IdealState
idealState, ExternalView
Set<String> onlineSegments) {
updateExplicitlySetTimeBoundary(idealState);
+ // Read the ZK metadata of the new segments in one batch
+ List<String> newSegments = new ArrayList<>();
+ List<String> newSegmentZKMetadataPaths = new ArrayList<>();
for (String segment : onlineSegments) {
// NOTE: Only update the segment end time when there are ONLINE
instances in the external view to prevent moving
// the time boundary before the new segment is picked up by the
servers
- Map<String, String> instanceStateMap = externalView.getStateMap(segment);
- if (instanceStateMap != null &&
instanceStateMap.containsValue(SegmentStateModel.ONLINE)) {
- _endTimeMsMap.computeIfAbsent(segment, k ->
extractEndTimeMsFromSegmentZKMetadataZNRecord(segment,
- _propertyStore.get(_segmentZKMetadataPathPrefix + segment, null,
AccessOption.PERSISTENT)));
+ if (!_endTimeMsMap.containsKey(segment)) {
+ Map<String, String> instanceStateMap =
externalView.getStateMap(segment);
+ if (instanceStateMap != null &&
instanceStateMap.containsValue(SegmentStateModel.ONLINE)) {
+ newSegments.add(segment);
+ newSegmentZKMetadataPaths.add(_segmentZKMetadataPathPrefix +
segment);
+ }
}
}
- _endTimeMsMap.keySet().retainAll(onlineSegments);
- updateTimeBoundaryInfo(getMaxEndTimeMs());
+ boolean recomputeMaxEndTime = false;
+ // NOTE: Removing a segment can only lower the max end time, so a full
scan is needed only when a removed segment
+ // holds the current max end time.
+ Iterator<Map.Entry<String, Long>> iterator =
_endTimeMsMap.entrySet().iterator();
+ while (iterator.hasNext()) {
+ Map.Entry<String, Long> entry = iterator.next();
+ if (!onlineSegments.contains(entry.getKey())) {
+ if (entry.getValue() >= _maxEndTimeMs) {
+ recomputeMaxEndTime = true;
+ }
+ iterator.remove();
+ }
+ }
+ if (recomputeMaxEndTime) {
+ _maxEndTimeMs = computeMaxEndTimeMs();
+ }
+ if (!newSegments.isEmpty()) {
+ List<ZNRecord> znRecords =
+ _propertyStore.get(newSegmentZKMetadataPaths, null,
AccessOption.PERSISTENT, false);
Review Comment:
Passing `false` changes the previous failure semantics: Helix returns null
for non-NONODE errors such as `OPERATIONTIMEOUT`, whereas the old single-node
read propagated these errors. During a replacement that removes the old offline
segments, the map and maximum have already been cleared here. If the
replacement metadata reads fail, the loop skips them and publishes a null time
boundary, causing hybrid routing to exclude the offline table and silently omit
historical rows. Returning normally also advances the routing entry's IS/EV
versions, so unchanged-version refreshes cannot retry until another assignment
change.
Could we pass `true` to preserve error propagation? Helix still returns null
for genuinely missing nodes with that setting. A regression test covering a
non-NONODE read failure during replacement would help verify that the published
boundary and retry eligibility are preserved.
##########
pinot-broker/src/test/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManagerTest.java:
##########
@@ -431,6 +434,133 @@ private void
assertPartitionHasNotOnlyDeferredSegments(String noReplicaSegment,
+
tablePartitionReplicatedServersInfo.getPartitionsWithOnlyDeferredSegments());
}
+ /// An assignment change that changes neither the segments, their partitions
nor their online servers must keep the
+ /// published objects, while a change of the online servers must only
replace the replicated servers info.
+ @Test
+ public void testRecomputeOnlyOnChange() {
+ ExternalView externalView = new ExternalView(OFFLINE_TABLE_NAME);
+ Map<String, Map<String, String>> segmentAssignment =
externalView.getRecord().getMapFields();
+ Set<String> onlineSegments = new HashSet<>();
+ // NOTE: Ideal state is not used in the current implementation.
+ IdealState idealState = new IdealState(OFFLINE_TABLE_NAME);
+
+ 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);
+
+ String segment0 = "recomputeSegment0";
+ String segment1 = "recomputeSegment1";
+ String segmentWithoutMetadata = "recomputeSegmentWithoutMetadata";
+ onlineSegments.add(segment0);
+ onlineSegments.add(segment1);
+ segmentAssignment.put(segment0, Map.of(SERVER_0, ONLINE, SERVER_1,
ONLINE));
+ segmentAssignment.put(segment1, Map.of(SERVER_0, ONLINE, SERVER_1,
ONLINE));
+ setSegmentZKMetadata(segment0, PARTITION_COLUMN_FUNC, NUM_PARTITIONS, 0,
0L);
+ setSegmentZKMetadata(segment1, PARTITION_COLUMN_FUNC, NUM_PARTITIONS, 1,
0L);
+ segmentZkMetadataFetcher.init(idealState, externalView, onlineSegments);
+ // A segment without ZK metadata is pulled again on every change, which
alone must not count as a change
+ onlineSegments.add(segmentWithoutMetadata);
+ segmentAssignment.put(segmentWithoutMetadata, Map.of(SERVER_0, ONLINE));
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ TablePartitionInfo tablePartitionInfo =
partitionMetadataManager.getTablePartitionInfo();
+ TablePartitionReplicatedServersInfo replicatedServersInfo =
+ partitionMetadataManager.getTablePartitionReplicatedServersInfo();
+ assertEquals(replicatedServersInfo.getSegmentsWithInvalidPartition(),
List.of(segmentWithoutMetadata));
+
+ // No-op change, with a fresh but equal instance state map and an
unrelated state change in the external view
+ segmentAssignment.put(segment0, new HashMap<>(Map.of(SERVER_0, ONLINE,
SERVER_1, ONLINE)));
+ segmentAssignment.put("segmentNotOnline", Map.of(SERVER_0, ONLINE));
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ assertSame(partitionMetadataManager.getTablePartitionInfo(),
tablePartitionInfo);
+
assertSame(partitionMetadataManager.getTablePartitionReplicatedServersInfo(),
replicatedServersInfo);
+
+ // An OFFLINE replica is not an online server
+ segmentAssignment.put(segment1, Map.of(SERVER_0, ONLINE, SERVER_1,
OFFLINE));
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ assertSame(partitionMetadataManager.getTablePartitionInfo(),
tablePartitionInfo);
+ replicatedServersInfo =
partitionMetadataManager.getTablePartitionReplicatedServersInfo();
+ TablePartitionReplicatedServersInfo.PartitionInfo[] partitionInfoMap =
replicatedServersInfo.getPartitionInfoMap();
+ assertEquals(partitionInfoMap[0]._fullyReplicatedServers, Set.of(SERVER_0,
SERVER_1));
+ assertEquals(partitionInfoMap[1]._fullyReplicatedServers,
Set.of(SERVER_0));
+
+ // Same online servers with a different state (CONSUMING instead of
ONLINE) is a no-op
+ segmentAssignment.put(segment1, Map.of(SERVER_0, CONSUMING, SERVER_1,
OFFLINE));
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ assertSame(partitionMetadataManager.getTablePartitionInfo(),
tablePartitionInfo);
+
assertSame(partitionMetadataManager.getTablePartitionReplicatedServersInfo(),
replicatedServersInfo);
+
+ // Removing a segment replaces both objects
+ onlineSegments.remove(segment1);
+ segmentAssignment.remove(segment1);
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ assertNotSame(partitionMetadataManager.getTablePartitionInfo(),
tablePartitionInfo);
+
assertNotSame(partitionMetadataManager.getTablePartitionReplicatedServersInfo(),
replicatedServersInfo);
+
assertEquals(partitionMetadataManager.getTablePartitionInfo().getSegmentsByPartition(),
+ List.of(List.of(segment0), List.of()));
+
assertNull(partitionMetadataManager.getTablePartitionReplicatedServersInfo().getPartitionInfoMap()[1]);
+ }
+
+ /// The partition info re-evaluates which segments are new only when it is
recomputed, so an assignment change that
+ /// changes nothing else must still recompute once a new segment excluded
from the partition info is no longer new.
+ @Test
+ public void testNewSegmentExpiresOnNoOpChange()
+ throws InterruptedException {
+ ExternalView externalView = new ExternalView(OFFLINE_TABLE_NAME);
+ Map<String, Map<String, String>> segmentAssignment =
externalView.getRecord().getMapFields();
+ Set<String> onlineSegments = new HashSet<>();
+ // NOTE: Ideal state is not used in the current implementation.
+ IdealState idealState = new IdealState(OFFLINE_TABLE_NAME);
+
+ long newSegmentExpirationMs = TimeUnit.MINUTES.toMillis(5);
+ SegmentPartitionMetadataManager partitionMetadataManager =
+ new SegmentPartitionMetadataManager(OFFLINE_TABLE_NAME,
PARTITION_COLUMN, PARTITION_COLUMN_FUNC, NUM_PARTITIONS,
+ newSegmentExpirationMs);
+ SegmentZkMetadataFetcher segmentZkMetadataFetcher =
+ new SegmentZkMetadataFetcher(OFFLINE_TABLE_NAME, _propertyStore);
+ segmentZkMetadataFetcher.register(partitionMetadataManager);
+
+ String oldSegment = "expiryOldSegment";
+ onlineSegments.add(oldSegment);
+ segmentAssignment.put(oldSegment, Map.of(SERVER_0, ONLINE, SERVER_1,
ONLINE));
+ setSegmentZKMetadata(oldSegment, PARTITION_COLUMN_FUNC, NUM_PARTITIONS, 0,
0L);
+ segmentZkMetadataFetcher.init(idealState, externalView, onlineSegments);
+
+ // A new segment missing one replica is excluded from the partition info.
It stops being new 1 second from now.
+ String newSegment = "expiryNewSegment";
+ long creationTimeMs = System.currentTimeMillis() - newSegmentExpirationMs
+ 1000;
+ onlineSegments.add(newSegment);
+ segmentAssignment.put(newSegment, Map.of(SERVER_0, ONLINE));
+ setSegmentZKMetadata(newSegment, PARTITION_COLUMN_FUNC, NUM_PARTITIONS, 0,
creationTimeMs);
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ TablePartitionReplicatedServersInfo replicatedServersInfo =
+ partitionMetadataManager.getTablePartitionReplicatedServersInfo();
+ TablePartitionReplicatedServersInfo.PartitionInfo partitionInfo =
replicatedServersInfo.getPartitionInfoMap()[0];
+ if (System.currentTimeMillis() - creationTimeMs <= newSegmentExpirationMs)
{
+ // Only assert the exclusion when the change above was processed while
the segment was still new
+ assertEquals(partitionInfo._segments, List.of(oldSegment));
+ assertEquals(partitionInfo._fullyReplicatedServers, Set.of(SERVER_0,
SERVER_1));
+ }
Review Comment:
Minor test concern: if the ZK write/read or scheduling before the first
assignment takes more than one second, the segment is already expired on its
initial build. This condition then skips the exclusion assertions, the wait is
skipped, and the test can pass even without the new expiry-triggered
recomputation branch. A controllable clock would let the test assert that the
segment is initially deferred, advance past expiration, and then verify that a
no-op assignment includes it.
--
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]
