This is an automated email from the ASF dual-hosted git repository.
shauryachats pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 26bf075d16c Skip missing consuming segment check for paused realtime
tables (#19310)
26bf075d16c is described below
commit 26bf075d16cfc6d80d5682589109f873a9c6dd96
Author: rohit <[email protected]>
AuthorDate: Sat Sep 12 04:00:33 2026 +0530
Skip missing consuming segment check for paused realtime tables (#19310)
SegmentStatusChecker invoked MissingConsumingSegmentFinder for every
enabled REALTIME table without checking whether ingestion was paused.
Pausing deliberately leaves a partition with no CONSUMING segment once
the current one commits, so the finder read that as a missing consumer
and the missingConsumingSegment* gauges reported a growing false
positive for as long as the pause lasted.
Gate the finder on the table pause state already computed in
updateSegmentMetrics. Because these gauges are last-write-wins with no
"unset" value, the paused branch actively resets them via a new
MissingConsumingSegmentFinder.resetMetrics, keeping ownership of the
gauge names with the class that emits them.
---
.../controller/helix/SegmentStatusChecker.java | 16 +++-
.../realtime/MissingConsumingSegmentFinder.java | 12 +++
.../controller/helix/SegmentStatusCheckerTest.java | 86 ++++++++++++++++++++++
3 files changed, 110 insertions(+), 4 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 cb27b48a9d7..423a3891bc6 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
@@ -300,7 +300,8 @@ public class SegmentStatusChecker extends
ControllerPeriodicTask<SegmentStatusCh
return false;
}
- if (PinotLLCRealtimeSegmentManager.isTablePaused(idealState)) {
+ boolean tablePaused =
PinotLLCRealtimeSegmentManager.isTablePaused(idealState);
+ if (tablePaused) {
context._pausedTables.add(tableNameWithType);
}
@@ -585,9 +586,16 @@ public class SegmentStatusChecker extends
ControllerPeriodicTask<SegmentStatusCh
numInvalidEndTime);
if (tableType == TableType.REALTIME && tableConfig != null) {
- List<StreamConfig> streamConfigs =
IngestionConfigUtils.getStreamConfigs(tableConfig);
- new MissingConsumingSegmentFinder(tableNameWithType, propertyStore,
_controllerMetrics,
- streamConfigs, idealState).findAndEmitMetrics(idealState);
+ if (tablePaused) {
+ // Ingestion is intentionally paused, so
PinotLLCRealtimeSegmentManager deliberately does not create a new
+ // CONSUMING segment after the current one commits.
MissingConsumingSegmentFinder would otherwise read that as
+ // a missing segment and alert for as long as the pause lasts.
+ MissingConsumingSegmentFinder.resetMetrics(tableNameWithType,
_controllerMetrics);
+ } else {
+ List<StreamConfig> streamConfigs =
IngestionConfigUtils.getStreamConfigs(tableConfig);
+ new MissingConsumingSegmentFinder(tableNameWithType, propertyStore,
_controllerMetrics,
+ streamConfigs, idealState).findAndEmitMetrics(idealState);
+ }
}
return true;
}
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/MissingConsumingSegmentFinder.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/MissingConsumingSegmentFinder.java
index 56d47cba7bc..13791ecdfc7 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/MissingConsumingSegmentFinder.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/MissingConsumingSegmentFinder.java
@@ -107,6 +107,18 @@ public class MissingConsumingSegmentFinder {
_streamPartitionMsgOffsetFactory = streamPartitionMsgOffsetFactory;
}
+ /// Zeroes out every gauge that [#findAndEmitMetrics] emits, for callers
that intentionally skip the search (e.g. a
+ /// table whose ingestion is paused, where having no CONSUMING segment is
expected rather than a defect). These
+ /// gauges are last-write-wins with no "unset" value, so a skipped table
would otherwise keep reporting whatever was
+ /// last written before the skip began. Kept here so the class that owns
these gauge names also owns their reset.
+ public static void resetMetrics(String realtimeTableName, ControllerMetrics
controllerMetrics) {
+ controllerMetrics.setValueOfTableGauge(realtimeTableName,
ControllerGauge.MISSING_CONSUMING_SEGMENT_TOTAL_COUNT, 0);
+ controllerMetrics.setValueOfTableGauge(realtimeTableName,
+ ControllerGauge.MISSING_CONSUMING_SEGMENT_NEW_PARTITION_COUNT, 0);
+ controllerMetrics.setValueOfTableGauge(realtimeTableName,
+ ControllerGauge.MISSING_CONSUMING_SEGMENT_MAX_DURATION_MINUTES, 0);
+ }
+
public void findAndEmitMetrics(IdealState idealState) {
MissingSegmentInfo info =
findMissingSegments(idealState.getRecord().getMapFields(), Instant.now());
_controllerMetrics.setValueOfTableGauge(_realtimeTableName,
ControllerGauge.MISSING_CONSUMING_SEGMENT_TOTAL_COUNT,
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 02e333eb52c..c9fb1f6c0a0 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
@@ -48,7 +48,9 @@ import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.api.resources.SegmentStatusInfo;
import org.apache.pinot.controller.api.resources.TableViews;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import
org.apache.pinot.controller.helix.core.realtime.PinotLLCRealtimeSegmentManager;
import org.apache.pinot.controller.util.TableSizeReader;
+import org.apache.pinot.spi.config.table.PauseState;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.TierConfig;
@@ -319,6 +321,90 @@ public class SegmentStatusCheckerTest {
ControllerGauge.MISSING_CONSUMING_SEGMENT_TOTAL_COUNT), 2);
}
+ /// When a table is paused via the pause/resume ingestion API,
PinotLLCRealtimeSegmentManager deliberately does not
+ /// create a new CONSUMING segment after the current one commits. Without
the pause gate,
+ /// MissingConsumingSegmentFinder would see the completed segments with no
CONSUMING replacement and flag them as
+ /// missing for as long as the pause lasts.
+ ///
+ /// The fixture matches [#realtimeBasicTest], which reports a
+ /// [ControllerGauge#MISSING_CONSUMING_SEGMENT_TOTAL_COUNT] of 2. The
checker is run twice against the same metrics
+ /// instance — once unpaused to establish that non-zero reading, then again
after pausing — so this covers the
+ /// actual production scenario: these gauges are last-write-wins, so pausing
must actively reset a previously
+ /// reported value rather than merely stop updating it.
+ @Test
+ public void realtimePausedTableHasNoMissingConsumingSegmentAlert() {
+ TableConfig tableConfig =
+ new
TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setTimeColumnName("timeColumn")
+ .setNumReplicas(3).setStreamConfigs(getStreamConfigMap()).build();
+
+ String seg1 = new LLCSegmentName(RAW_TABLE_NAME, 1, 0,
System.currentTimeMillis()).getSegmentName();
+ String seg2 = new LLCSegmentName(RAW_TABLE_NAME, 1, 1,
System.currentTimeMillis()).getSegmentName();
+ String seg3 = new LLCSegmentName(RAW_TABLE_NAME, 2, 1,
System.currentTimeMillis()).getSegmentName();
+ IdealState idealState = new IdealState(REALTIME_TABLE_NAME);
+ idealState.setPartitionState(seg1, "pinot1", "ONLINE");
+ idealState.setPartitionState(seg1, "pinot2", "ONLINE");
+ idealState.setPartitionState(seg1, "pinot3", "ONLINE");
+
+ idealState.setPartitionState(seg2, "pinot1", "ONLINE");
+ idealState.setPartitionState(seg2, "pinot2", "ONLINE");
+ idealState.setPartitionState(seg2, "pinot3", "ONLINE");
+
+ idealState.setPartitionState(seg3, "pinot1", "CONSUMING");
+ idealState.setPartitionState(seg3, "pinot2", "CONSUMING");
+ idealState.setPartitionState(seg3, "pinot3", "OFFLINE");
+ idealState.setReplicas("3");
+ idealState.setRebalanceMode(IdealState.RebalanceMode.CUSTOMIZED);
+
+ ExternalView externalView = new ExternalView(REALTIME_TABLE_NAME);
+ externalView.setState(seg1, "pinot1", "ONLINE");
+ externalView.setState(seg1, "pinot2", "ONLINE");
+ externalView.setState(seg1, "pinot3", "ONLINE");
+
+ externalView.setState(seg2, "pinot1", "CONSUMING");
+ externalView.setState(seg2, "pinot2", "ONLINE");
+ externalView.setState(seg2, "pinot3", "CONSUMING");
+
+ externalView.setState(seg3, "pinot1", "CONSUMING");
+ externalView.setState(seg3, "pinot2", "CONSUMING");
+ externalView.setState(seg3, "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);
+ SegmentZKMetadata committedSegmentZKMetadata =
mockCommittedSegmentZKMetadata();
+ SegmentZKMetadata consumingSegmentZKMetadata =
mockConsumingSegmentZKMetadata(11111L);
+ mockSegmentsZKMetadata(resourceManager, REALTIME_TABLE_NAME,
+ Map.of(seg1, committedSegmentZKMetadata, seg2,
committedSegmentZKMetadata, seg3, consumingSegmentZKMetadata));
+
+ // Stubbed exactly as in realtimeBasicTest so that
MissingConsumingSegmentFinder computes a non-zero count.
+ 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);
+
+ // Not yet paused: the finder runs and reports the missing consuming
segments.
+ runSegmentStatusChecker(resourceManager, 0);
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
REALTIME_TABLE_NAME,
+ ControllerGauge.MISSING_CONSUMING_SEGMENT_TOTAL_COUNT), 2);
+
+ // Mirrors how PinotLLCRealtimeSegmentManager#updatePauseStateInIdealState
records an administrative pause, and how
+ // #isTablePaused reads it back off the ideal state record.
+
idealState.getRecord().setSimpleField(PinotLLCRealtimeSegmentManager.PAUSE_STATE,
+ new PauseState(true, PauseState.ReasonCode.ADMINISTRATIVE, "paused for
test", "0", null).toJsonString());
+
+ runSegmentStatusChecker(resourceManager, 0);
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
REALTIME_TABLE_NAME,
+ ControllerGauge.MISSING_CONSUMING_SEGMENT_TOTAL_COUNT), 0);
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
REALTIME_TABLE_NAME,
+ ControllerGauge.MISSING_CONSUMING_SEGMENT_NEW_PARTITION_COUNT), 0);
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
REALTIME_TABLE_NAME,
+ ControllerGauge.MISSING_CONSUMING_SEGMENT_MAX_DURATION_MINUTES), 0);
+ }
+
@Test
public void realtimeMutableSegmentHasLessReplicaTest() {
TableConfig tableConfig =
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
