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);