gortiz commented on code in PR #19783:
URL: https://github.com/apache/pinot/pull/19783#discussion_r4258992480
##########
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:
Good catch, thanks. The batched read now passes `throwException=true`, so a
ZK error other than NONODE propagates and the routing entry keeps its IS/EV
versions. I also moved the read before any state change, so a failed read
leaves the end time map, the max end time and the published boundary as they
were. `testReadFailureDuringReplacementKeepsTimeBoundary` covers an
OPERATIONTIMEOUT during a replacement and the retry after it.
##########
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:
Done. The manager now takes a `Clock`, and the test uses `TestClock`. It
always asserts the initial exclusion, checks that a no-op change keeps the
object while the segment is new, then advances the clock and checks the
recompute.
--
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]
