yashmayya commented on code in PR #19252:
URL: https://github.com/apache/pinot/pull/19252#discussion_r3799115547
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/SegmentStatusCheckerTest.java:
##########
@@ -626,6 +671,198 @@ public void
realtimeCommittingSegmentWithinGraceNotUnderReplicated() {
ControllerGauge.SEGMENTS_WITH_LESS_REPLICAS), 0);
}
+ /// When a segment's znode stat is unavailable, the grace window falls back
to the metadata's creation time. A freshly
+ /// created CONSUMING segment (mtime == creation time) must therefore still
be graced, so a lost stat cannot turn into
+ /// a false under-replication alert.
+ @Test
+ public void
realtimeConsumingSegmentWithoutZNodeStatFallsBackToCreationTime() {
+ TableConfig tableConfig =
+ new
TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setTimeColumnName("timeColumn")
+ .setNumReplicas(3).setStreamConfigs(getStreamConfigMap()).build();
+
+ String seg = new LLCSegmentName(RAW_TABLE_NAME, 1, 5,
System.currentTimeMillis()).getSegmentName();
+ IdealState idealState = new IdealState(REALTIME_TABLE_NAME);
+ idealState.setPartitionState(seg, "pinot1", "CONSUMING");
+ idealState.setPartitionState(seg, "pinot2", "CONSUMING");
+ idealState.setPartitionState(seg, "pinot3", "CONSUMING");
+ idealState.setReplicas("3");
+ idealState.setRebalanceMode(IdealState.RebalanceMode.CUSTOMIZED);
+
+ // Just created: only 1 of 3 replicas has started consuming.
+ ExternalView externalView = new ExternalView(REALTIME_TABLE_NAME);
+ externalView.setState(seg, "pinot1", "CONSUMING");
+ externalView.setState(seg, "pinot2", "OFFLINE");
+ externalView.setState(seg, "pinot3", "OFFLINE");
+
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+
when(resourceManager.getHelixInstanceConfig(any())).thenReturn(newQuerableInstanceConfig("any"));
+
when(resourceManager.getTableConfig(REALTIME_TABLE_NAME)).thenReturn(tableConfig);
+
when(resourceManager.getAllTables()).thenReturn(List.of(REALTIME_TABLE_NAME));
+
when(resourceManager.getTableIdealState(REALTIME_TABLE_NAME)).thenReturn(idealState);
+
when(resourceManager.getTableExternalView(REALTIME_TABLE_NAME)).thenReturn(externalView);
+ // Creation time is now, and no znode stat is available -> the fallback
must keep the segment within the grace
+ // window
+ SegmentZKMetadata consumingSegmentZKMetadata =
mockConsumingSegmentZKMetadata(System.currentTimeMillis());
+ mockSegmentsZKMetadata(resourceManager, REALTIME_TABLE_NAME, Map.of(seg,
consumingSegmentZKMetadata));
+
+ ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
+ when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
+ ZNRecord znRecord = new ZNRecord("0");
+ znRecord.setSimpleField(CommonConstants.Segment.Realtime.END_OFFSET,
"10000");
+ when(propertyStore.get(anyString(), any(), anyInt())).thenReturn(znRecord);
+
+ // 1h grace window; the segment was just created, so it must be skipped
and the table stays fully replicated.
+ runSegmentStatusChecker(resourceManager, 3600);
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
REALTIME_TABLE_NAME,
+ ControllerGauge.PERCENT_OF_REPLICAS), 100);
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
REALTIME_TABLE_NAME,
+ ControllerGauge.SEGMENTS_WITH_LESS_REPLICAS), 0);
+ }
+
+ /// The whole table's metadata and znode stats come back from batched reads,
so each segment must be matched to its
+ /// own metadata (by name) and its own stat (by position) rather than to
whichever entry happens to sit at its index.
+ /// Every segment is under-replicated and carries a distinct size, exactly
one segment was pushed recently enough to
+ /// be graced, and the last segment is the only one with 4 replicas so it
alone determines PERCENT_OF_REPLICAS.
+ /// Together the gauges pin that every segment was examined under its own
name, that the grace window applied to
+ /// exactly the segment whose znode stat is recent, and that the sizes
accumulate. One segment has no ZK metadata,
+ /// which shifts the alignment if the pairing gets it wrong.
+ @Test
+ public void segmentsStayAlignedWithTheirBatchedMetadata() {
+ TableConfig tableConfig =
+ new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(2).build();
+
+ int numSegments = 7;
+ // Distinct positions, none of them the last segment, which carries its
own marker below
+ int segmentWithoutZKMetadata = 3;
+ int recentlyPushedSegment = 1;
+ // The last segment is the marker that pins name-to-metadata pairing: it
is the only one whose replica ratio is 1/4
+ // rather than 1/2, so PERCENT_OF_REPLICAS drops to 25 only if this exact
segment was examined.
+ int lowReplicaSegment = numSegments - 1;
+ long oldPushTimeMs = 11111L;
+
+ IdealState idealState = new IdealState(OFFLINE_TABLE_NAME);
+ ExternalView externalView = new ExternalView(OFFLINE_TABLE_NAME);
+ Map<String, SegmentZKMetadata> segmentZKMetadataMap = new HashMap<>();
+ Map<String, Long> segmentZNodeMTimesMs = new HashMap<>();
+ long expectedTableCompressedSize = 0;
+ for (int i = 0; i < numSegments; i++) {
+ String segment = "myTable_" + i;
+ int numReplicas = i == lowReplicaSegment ? 4 : 2;
+ // Every segment is under-replicated, so any segment that is examined
and not graced must be counted
+ for (int replica = 1; replica <= numReplicas; replica++) {
+ idealState.setPartitionState(segment, "pinot" + replica, "ONLINE");
+ externalView.setState(segment, "pinot" + replica, replica == 1 ?
"ONLINE" : "OFFLINE");
+ }
+ if (i == segmentWithoutZKMetadata) {
+ continue;
+ }
+ // Distinct size per segment so that the total pins which metadata was
attributed to which segment
+ long sizeInBytes = 1000L + i;
+ segmentZKMetadataMap.put(segment,
mockPushedSegmentZKMetadata(sizeInBytes, oldPushTimeMs));
+ segmentZNodeMTimesMs.put(segment, i == recentlyPushedSegment ?
System.currentTimeMillis() : oldPushTimeMs);
+ expectedTableCompressedSize += sizeInBytes;
+ }
+ idealState.setReplicas("2");
+ idealState.setRebalanceMode(IdealState.RebalanceMode.CUSTOMIZED);
+
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+
when(resourceManager.getHelixInstanceConfig(any())).thenReturn(newQuerableInstanceConfig("any"));
+
when(resourceManager.getAllTables()).thenReturn(List.of(OFFLINE_TABLE_NAME));
+
when(resourceManager.getTableConfig(OFFLINE_TABLE_NAME)).thenReturn(tableConfig);
+
when(resourceManager.getTableIdealState(OFFLINE_TABLE_NAME)).thenReturn(idealState);
+
when(resourceManager.getTableExternalView(OFFLINE_TABLE_NAME)).thenReturn(externalView);
+ mockSegmentsZKMetadata(resourceManager, OFFLINE_TABLE_NAME,
segmentZKMetadataMap, segmentZNodeMTimesMs);
+
+ ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
+ when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
+
+ // 10min grace window, so only the recently pushed segment is skipped
+ runSegmentStatusChecker(resourceManager, 600, mock(TableSizeReader.class));
+
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
OFFLINE_TABLE_NAME,
+ ControllerGauge.SEGMENT_COUNT), numSegments);
+ // Every segment except the one without ZK metadata and the graced one
must be counted. A segment paired with the
+ // wrong metadata or the wrong znode stat drops out of, or into, this
count.
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
OFFLINE_TABLE_NAME,
+ ControllerGauge.SEGMENTS_WITH_LESS_REPLICAS), numSegments - 2);
+ // Only the last segment has a 1-of-4 ratio, so this is 25 only if that
segment was examined under its own name
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
OFFLINE_TABLE_NAME,
+ ControllerGauge.PERCENT_OF_REPLICAS), 25);
+ // The sizes accumulate over exactly the segments that have metadata
(including the graced one, whose size is
+ // counted before the grace check)
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
OFFLINE_TABLE_NAME,
+ ControllerGauge.TABLE_COMPRESSED_SIZE), expectedTableCompressedSize);
+
+ // A single batched read must cover the whole table, metadata and znode
stats together: more than one call means the
+ // per-segment reads crept back in some form
+ ArgumentCaptor<List<String>> segmentNamesCaptor =
ArgumentCaptor.forClass(List.class);
+
verify(resourceManager).getSegmentsZKMetadataForSegmentNames(eq(OFFLINE_TABLE_NAME),
segmentNamesCaptor.capture(),
+ any());
+ verify(propertyStore, never()).getStats(any(), anyInt());
+ List<String> requestedSegments = segmentNamesCaptor.getValue();
+ assertEquals(requestedSegments.size(), numSegments);
+ assertEquals(new HashSet<>(requestedSegments),
idealState.getPartitionSet());
+ }
+
+ /// When not a single segment's ZK metadata can be read the table's gauges
must be left alone rather than reset to
+ /// all-green values, because an all-green gauge silences the alerts that a
stale one would still fire. Regression
+ /// test for a whole-table ZK read failure being reported as a perfectly
healthy table.
+ @Test
+ public void tableWithoutAnyReadableSegmentZKMetadataKeepsItsGauges() {
Review Comment:
Is this testing the opposite of `noSegmentZKMetadataTest`? Should that other
test be updated / deleted?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java:
##########
@@ -382,10 +383,25 @@ private boolean updateSegmentMetrics(String
tableNameWithType, TableConfig table
List<String> unavailableSegmentsByState = new ArrayList<>();
List<String> unavailableSegmentsByInstance = new ArrayList<>();
- for (String segment : segments) {
+ // The segment ZK metadata and the znode stats are read in a single
batched request
+ List<String> segmentsToCheck = new ArrayList<>(segments);
+ List<Stat> segmentStats = new ArrayList<>(numSegments);
+ List<SegmentZKMetadata> segmentsZKMetadata =
Review Comment:
Should we consider chunking this batch request instead of having an
unbounded batch size? For tables that are super large (1M+ segments), the
unbounded approach could cause heap spikes or even OOM on the controller, no?
Maybe something like 5k-10k per batch will allow us to retain the speedup
benefit but also cap the amount of heap used.
##########
pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java:
##########
@@ -639,8 +640,36 @@ public static Schema
getTableSchema(ZkHelixPropertyStore<ZNRecord> propertyStore
return getTableSchema(propertyStore, tableConfig.getTableName());
}
+ /// Reads the ZK metadata of the named segments of the given table in a
single batched request, and returns it
+ /// index-aligned with `segmentNames`. An entry is `null` when the segment's
znode could not be read, either because
+ /// it does not exist or because the read failed; the two are not
distinguished.
+ ///
+ /// When `stats` is non-null it is filled with the [Stat] of each segment's
znode, also index-aligned with
+ /// `segmentNames`, and `null` wherever the record is `null`.
+ public static List<SegmentZKMetadata>
getSegmentsZKMetadata(ZkHelixPropertyStore<ZNRecord> propertyStore,
Review Comment:
`getSegmentsZKMetadata` has retries but this method does not, I'm guessing
that wasn't intentional?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1051,6 +1051,15 @@ public List<SegmentZKMetadata>
getSegmentsZKMetadata(String tableNameWithType) {
return ZKMetadataProvider.getSegmentsZKMetadata(_propertyStore,
tableNameWithType);
}
+ /// Reads the ZK metadata of the named segments in a single batched request,
index-aligned with `segmentNames` and
+ /// holding `null` for segments whose znode could not be read. See
+ /// [ZKMetadataProvider#getSegmentsZKMetadata(ZkHelixPropertyStore, String,
List, List)] for the full contract,
+ /// including the `stats` out-parameter that carries the znodes'
modification times.
+ public List<SegmentZKMetadata> getSegmentsZKMetadataForSegmentNames(String
tableNameWithType,
Review Comment:
nit: this uses a new method name but the new method added in
`ZKMetadataProvider` is an overload. Let's stick to one convention (I'm fine
with either, so up to you).
##########
pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java:
##########
@@ -639,8 +640,36 @@ public static Schema
getTableSchema(ZkHelixPropertyStore<ZNRecord> propertyStore
return getTableSchema(propertyStore, tableConfig.getTableName());
}
+ /// Reads the ZK metadata of the named segments of the given table in a
single batched request, and returns it
+ /// index-aligned with `segmentNames`. An entry is `null` when the segment's
znode could not be read, either because
+ /// it does not exist or because the read failed; the two are not
distinguished.
+ ///
+ /// When `stats` is non-null it is filled with the [Stat] of each segment's
znode, also index-aligned with
+ /// `segmentNames`, and `null` wherever the record is `null`.
+ public static List<SegmentZKMetadata>
getSegmentsZKMetadata(ZkHelixPropertyStore<ZNRecord> propertyStore,
+ String tableNameWithType, List<String> segmentNames, @Nullable
List<Stat> stats) {
+ int numSegments = segmentNames.size();
+ List<String> paths = new ArrayList<>(numSegments);
+ for (String segmentName : segmentNames) {
+ paths.add(constructPropertyStorePathForSegment(tableNameWithType,
segmentName));
+ }
+ List<ZNRecord> znRecords = propertyStore.get(paths, stats,
AccessOption.PERSISTENT, false);
Review Comment:
Why are we passing `false` for the `throwException` parameter? Won't passing
`true` instead give exactly the semantics the checker wants: deleted segment →
`null`, failed read → exception, and `processTable` already has a catch for the
latter?
--
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]
