From f986a9238da94e6dfe222304ee6d57662397ccbb Mon Sep 17 00:00:00 2001 From: Krishan Goyal Date: Wed, 8 Jul 2026 13:49:50 +0530 Subject: [PATCH] Skip recently committed (COMMITTING) segments in SegmentStatusChecker SegmentStatusChecker's "skip recently created/pushed segments" logic predated the pauseless COMMITTING status and only special-cased IN_PROGRESS. A COMMITTING segment (consumed but still being built and uploaded by the server) therefore: - fell into the push-time branch, and since realtime segments have no push time, was never skipped during the wait-for-push grace period, so it was counted as under-replicated/offline while servers were still loading it; and - was subjected to start/end time-range validation even though start and end time are only finalized once the segment becomes DONE, so it was flagged with an invalid time range. Both checks now key off Status.isCompleted() (DONE/UPLOADED), so IN_PROGRESS and COMMITTING segments are handled the same way, removing the false alerts and log noise for pauseless tables. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../helix/SegmentStatusChecker.java | 19 +++-- .../helix/SegmentStatusCheckerTest.java | 71 +++++++++++++++++++ 2 files changed, 83 insertions(+), 7 deletions(-) diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java index 2d06ea9d976b..12870c964a9e 100644 --- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java +++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java @@ -402,21 +402,26 @@ private void updateSegmentMetrics(String tableNameWithType, TableConfig tableCon } // NOTE: We want to skip segments that are just created/pushed to avoid false alerts because it is expected for - // servers to take some time to load them. For consuming (IN_PROGRESS) segments, we use creation time from - // the ZK metadata; for pushed segments, we use push time from the ZK metadata. Both of them are the time - // when segment is newly created. For committed segments from real-time table, push time doesn't exist, and - // creationTimeMs will be Long.MIN_VALUE, which is fine because we want to include them in the check. + // servers to take some time to load them. For segments that are not yet completed (still consuming + // IN_PROGRESS segments, and COMMITTING segments of pauseless tables that have finished consuming but are + // still being built and uploaded by the server), we use creation time from the ZK metadata; for pushed + // segments, we use push time from the ZK metadata. Both of them are the time when segment is newly created. + // For committed (DONE) segments from real-time table, push time doesn't exist, and creationTimeMs will be + // Long.MIN_VALUE, which is fine because we want to include them in the check. // The comparison uses evSnapshotTimestamp instead of System.currentTimeMillis() because for large tables // with many segments, the status check can take several minutes. A segment updated after // the EV snapshot was taken but before this individual segment check runs could be incorrectly flagged as // OFFLINE when using current time. - long creationTimeMs = segmentZKMetadata.getStatus() == Status.IN_PROGRESS ? segmentZKMetadata.getCreationTime() - : segmentZKMetadata.getPushTime(); + Status segmentStatus = segmentZKMetadata.getStatus(); + long creationTimeMs = + segmentStatus.isCompleted() ? segmentZKMetadata.getPushTime() : segmentZKMetadata.getCreationTime(); if (creationTimeMs > evSnapshotTimestamp - _waitForPushTimeSeconds * 1000L) { continue; } - if (segmentZKMetadata.getStatus() != Status.IN_PROGRESS) { + // Only completed (DONE/UPLOADED) segments have finalized start/end time in the ZK metadata. Consuming + // (IN_PROGRESS) and COMMITTING segments do not, so skip the time range validation for them to avoid false alerts. + if (segmentStatus.isCompleted()) { if (!TimeUtils.timeValueInValidRange(segmentZKMetadata.getStartTimeMs())) { segmentsInvalidStartTime.add(segment); } diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/SegmentStatusCheckerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/SegmentStatusCheckerTest.java index 0066e482c28e..774d43e79534 100644 --- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/SegmentStatusCheckerTest.java +++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/SegmentStatusCheckerTest.java @@ -473,6 +473,69 @@ public void realtimeImmutableSegmentHasLessReplicaTest() { ControllerGauge.MISSING_CONSUMING_SEGMENT_TOTAL_COUNT), 2); } + /** + * COMMITTING segments of pauseless tables have been consumed but are still being built and uploaded by the server, + * so they should be treated like consuming (IN_PROGRESS) segments rather than committed ones: a recently created one + * is skipped from the checks during the wait-for-push grace period, and none of them is flagged for an invalid time + * range since their time range is only finalized once they become DONE. + */ + @Test + public void realtimeCommittingSegmentTest() { + TableConfig tableConfig = + new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setTimeColumnName("timeColumn") + .setNumReplicas(3).setStreamConfigs(getStreamConfigMap()).build(); + + long now = System.currentTimeMillis(); + String recentCommittingSegment = new LLCSegmentName(RAW_TABLE_NAME, 1, 0, now).getSegmentName(); + String oldCommittingSegment = new LLCSegmentName(RAW_TABLE_NAME, 2, 0, now).getSegmentName(); + + // Both segments are ONLINE on all three replicas in the ideal state, but only one replica is ONLINE in the + // external view, mimicking replicas that are still building/downloading the freshly committed segment. + IdealState idealState = new IdealState(REALTIME_TABLE_NAME); + ExternalView externalView = new ExternalView(REALTIME_TABLE_NAME); + for (String segment : List.of(recentCommittingSegment, oldCommittingSegment)) { + idealState.setPartitionState(segment, "pinot1", "ONLINE"); + idealState.setPartitionState(segment, "pinot2", "ONLINE"); + idealState.setPartitionState(segment, "pinot3", "ONLINE"); + externalView.setState(segment, "pinot1", "ONLINE"); + externalView.setState(segment, "pinot2", "OFFLINE"); + externalView.setState(segment, "pinot3", "OFFLINE"); + } + idealState.setReplicas("3"); + idealState.setRebalanceMode(IdealState.RebalanceMode.CUSTOMIZED); + + 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); + SegmentZKMetadata recentCommittingSegmentZKMetadata = mockCommittingSegmentZKMetadata(now); + when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME, recentCommittingSegment)).thenReturn( + recentCommittingSegmentZKMetadata); + SegmentZKMetadata oldCommittingSegmentZKMetadata = mockCommittingSegmentZKMetadata(11111L); + when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME, oldCommittingSegment)).thenReturn( + oldCommittingSegmentZKMetadata); + + ZkHelixPropertyStore 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); + + // Use a large wait-for-push window so the recently created COMMITTING segment is skipped while the old one is not. + runSegmentStatusChecker(resourceManager, 3600); + + // Only the old COMMITTING segment is checked for replica availability; the recent one is skipped entirely. + assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics, REALTIME_TABLE_NAME, + ControllerGauge.SEGMENTS_WITH_LESS_REPLICAS), 1); + // COMMITTING segments have not finalized their time range yet, so neither should be flagged for invalid time. + assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics, REALTIME_TABLE_NAME, + ControllerGauge.SEGMENTS_WITH_INVALID_START_TIME), 0); + assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics, REALTIME_TABLE_NAME, + ControllerGauge.SEGMENTS_WITH_INVALID_END_TIME), 0); + } + private Map getStreamConfigMap() { return Map.of("streamType", "kafka", "stream.kafka.topic.name", "test", "stream.kafka.decoder.class.name", "org.apache.pinot.plugin.stream.kafka.KafkaAvroMessageDecoder", "stream.kafka.consumer.factory.class.name", @@ -495,6 +558,14 @@ private SegmentZKMetadata mockConsumingSegmentZKMetadata(long creationTimeMs) { return segmentZKMetadata; } + private SegmentZKMetadata mockCommittingSegmentZKMetadata(long creationTimeMs) { + SegmentZKMetadata segmentZKMetadata = mock(SegmentZKMetadata.class); + when(segmentZKMetadata.getStatus()).thenReturn(Status.COMMITTING); + when(segmentZKMetadata.getSizeInBytes()).thenReturn(-1L); + when(segmentZKMetadata.getCreationTime()).thenReturn(creationTimeMs); + return segmentZKMetadata; + } + @Test public void missingEVPartitionTest() { IdealState idealState = new IdealState(OFFLINE_TABLE_NAME);