This is an automated email from the ASF dual-hosted git repository.
noob-se7en 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 2b8007eb634 Mitigate stale realtime segmentStoppedConsuming
notification (#18914)
2b8007eb634 is described below
commit 2b8007eb6343dad88edae2173b12e1ec43dc79ea
Author: Goutam Adwant <[email protected]>
AuthorDate: Mon Jul 6 23:43:36 2026 -0700
Mitigate stale realtime segmentStoppedConsuming notification (#18914)
* Fix realtime segment cleanup after ERROR transition
* Scope stale stop-consumed mitigation
---
.../realtime/RealtimeSegmentDataManager.java | 38 ++++++++++++--
.../realtime/RealtimeSegmentDataManagerTest.java | 60 +++++++++++++++++++++-
2 files changed, 94 insertions(+), 4 deletions(-)
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
index b060f2448a5..47695c74ec2 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
@@ -1505,7 +1505,8 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
Thread.sleep(SegmentCompletionProtocol.MAX_HOLD_TIME_MS);
}
- private static class ConsumptionStopIndicator {
+ @VisibleForTesting
+ static class ConsumptionStopIndicator {
final StreamPartitionMsgOffset _offset;
final String _segmentName;
final String _instanceId;
@@ -1539,7 +1540,7 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
ConsumptionStopIndicator indicator = new
ConsumptionStopIndicator(_currentOffset,
_segmentNameStr, _instanceId, _protocolHandler, reason,
_segmentLogger);
do {
- SegmentCompletionProtocol.Response response =
indicator.postSegmentStoppedConsuming();
+ SegmentCompletionProtocol.Response response =
postSegmentStoppedConsuming(indicator);
if (response.getStatus() ==
SegmentCompletionProtocol.ControllerResponseStatus.PROCESSED) {
break;
}
@@ -1548,6 +1549,37 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
} while (!_shouldStop);
}
+ @VisibleForTesting
+ void postStopConsumedMsgForInitializationError() {
+ if (hasDifferentSegmentDataManagerRegistered()) {
+ _segmentLogger.info(
+ "Skip segmentStoppedConsuming for segment: {}, another segment data
manager is already registered",
+ _segmentNameStr);
+ return;
+ }
+ postStopConsumedMsg("Consuming segment initialization error");
+ }
+
+ @VisibleForTesting
+ SegmentCompletionProtocol.Response
postSegmentStoppedConsuming(ConsumptionStopIndicator indicator) {
+ return indicator.postSegmentStoppedConsuming();
+ }
+
+ /**
+ * Returns true when another manager is currently registered for this
segment.
+ */
+ @VisibleForTesting
+ boolean hasDifferentSegmentDataManagerRegistered() {
+ if (_realtimeTableDataManager == null) {
+ return false;
+ }
+ SegmentDataManager segmentDataManager =
_realtimeTableDataManager.getSegmentDataManager(_segmentNameStr);
+ if (segmentDataManager == null) {
+ return false;
+ }
+ return segmentDataManager != this;
+ }
+
public StreamMetadataProvider getPartitionMetadataProvider() {
return _partitionMetadataProvider;
}
@@ -1985,7 +2017,7 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
// we are about to receive an ERROR->OFFLINE state transition once we
call
// postSegmentStoppedConsuming() method.
Uninterruptibles.sleepUninterruptibly(30, TimeUnit.SECONDS);
- postStopConsumedMsg("Consuming segment initialization error");
+ postStopConsumedMsgForInitializationError();
}).start();
throw t;
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
index 92a6dad4763..95a64108511 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
@@ -46,6 +46,7 @@ import
org.apache.pinot.core.data.manager.provider.TableDataManagerProvider;
import org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils;
import
org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConsumerFactory;
import org.apache.pinot.core.realtime.impl.fakestream.FakeStreamMessageDecoder;
+import org.apache.pinot.segment.local.data.manager.SegmentDataManager;
import org.apache.pinot.segment.local.data.manager.TableDataManager;
import
org.apache.pinot.segment.local.realtime.impl.RealtimeSegmentStatsHistory;
import org.apache.pinot.segment.local.segment.creator.Fixtures;
@@ -139,6 +140,47 @@ public class RealtimeSegmentDataManagerTest {
return createFakeSegmentManager(false, new TimeSupplier(), null, null,
null);
}
+ @Test
+ public void
testInitializationErrorStopMsgSkippedWhenDifferentSegmentManagerRegistered()
+ throws Exception {
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
createFakeSegmentManager()) {
+ RealtimeTableDataManager tableDataManager =
segmentDataManager.getTableDataManager();
+ SegmentDataManager registeredSegmentDataManager =
mock(SegmentDataManager.class);
+
when(tableDataManager.getSegmentDataManager(SEGMENT_NAME_STR)).thenReturn(registeredSegmentDataManager);
+
+ segmentDataManager.postStopConsumedMsgForInitializationError();
+
+ Assert.assertFalse(segmentDataManager._postConsumeStoppedCalled);
+ }
+ }
+
+ @Test
+ public void
testInitializationErrorStopMsgSentWhenCurrentSegmentManagerRegistered()
+ throws Exception {
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
createFakeSegmentManager()) {
+ RealtimeTableDataManager tableDataManager =
segmentDataManager.getTableDataManager();
+
when(tableDataManager.getSegmentDataManager(SEGMENT_NAME_STR)).thenReturn(segmentDataManager);
+
+ segmentDataManager.postStopConsumedMsgForInitializationError();
+
+ Assert.assertTrue(segmentDataManager._postConsumeStoppedCalled);
+ }
+ }
+
+ @Test
+ public void testPostStopConsumedMsgDoesNotCheckRegisteredSegmentManager()
+ throws Exception {
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
createFakeSegmentManager()) {
+ RealtimeTableDataManager tableDataManager =
segmentDataManager.getTableDataManager();
+ SegmentDataManager registeredSegmentDataManager =
mock(SegmentDataManager.class);
+
when(tableDataManager.getSegmentDataManager(SEGMENT_NAME_STR)).thenReturn(registeredSegmentDataManager);
+
+ segmentDataManager.invokePostStopConsumedMsg("consumer error");
+
+ Assert.assertTrue(segmentDataManager._postConsumeStoppedCalled);
+ }
+ }
+
private FakeRealtimeSegmentDataManager createFakeSegmentManager(boolean
noUpsert, TimeSupplier timeSupplier,
@Nullable String maxRows, @Nullable String maxDuration, @Nullable
TableConfig tableConfig)
throws Exception {
@@ -1372,6 +1414,7 @@ public class RealtimeSegmentDataManagerTest {
new IndexLoadingConfig(makeInstanceDataManagerConfig(),
tableConfig), schema, llcSegmentName,
consumerCoordinatorMap.get(llcSegmentName.getPartitionGroupId()),
serverMetrics, null, null,
() -> true);
+ _tableDataManager = realtimeTableDataManager;
_state = RealtimeSegmentDataManager.class.getDeclaredField("_state");
_state.setAccessible(true);
_shouldStop =
RealtimeSegmentDataManager.class.getDeclaredField("_shouldStop");
@@ -1386,7 +1429,10 @@ public class RealtimeSegmentDataManagerTest {
_streamMsgOffsetFactory.setAccessible(true);
_streamMsgOffsetFactory.set(this, new LongMsgOffsetFactory());
_timeSupplier = timeSupplier;
- _tableDataManager = realtimeTableDataManager;
+ }
+
+ public RealtimeTableDataManager getTableDataManager() {
+ return _tableDataManager;
}
public String getStopReason() {
@@ -1461,6 +1507,18 @@ public class RealtimeSegmentDataManagerTest {
_postConsumeStoppedCalled = true;
}
+ public void invokePostStopConsumedMsg(String reason) {
+ super.postStopConsumedMsg(reason);
+ }
+
+ @Override
+ SegmentCompletionProtocol.Response
postSegmentStoppedConsuming(ConsumptionStopIndicator indicator) {
+ _postConsumeStoppedCalled = true;
+ return new SegmentCompletionProtocol.Response(
+ new SegmentCompletionProtocol.Response.Params().withStatus(
+ SegmentCompletionProtocol.ControllerResponseStatus.PROCESSED));
+ }
+
@Override
protected void notifySegmentBuildFailedWithDeterministicError() {
_notifySegmentBuildFailedWithDeterministicErrorCalled = true;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
