This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch codex/consensus-poison-frontier in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 82e3dbc39fcd0939ee65a9f8711ae39f8f01ac10 Author: Caideyipi <[email protected]> AuthorDate: Mon Oct 12 19:38:46 2026 +0800 [Subscription] Advance consensus progress when discarding poisoned events --- .../consensus/ConsensusPrefetchingQueue.java | 15 +- .../consensus/ConsensusPrefetchingQueueTest.java | 172 +++++++++++++++++++++ 2 files changed, 184 insertions(+), 3 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java index dd5d2040009..f655afd5aed 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java @@ -3217,7 +3217,7 @@ public class ConsensusPrefetchingQueue { } ev.nack(); nacked.set(true); - if (ev.isPoisoned()) { + if (ev.isPoisoned() && commitPoisonedEvent(ev)) { LOGGER.error( DataNodePipeMessages .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_POISON_MESSAGE_DETECTED_NACKCOUNT_3A9255FB, @@ -3255,7 +3255,7 @@ public class ConsensusPrefetchingQueue { ev.nack(); nacked.set(true); - if (ev.isPoisoned()) { + if (ev.isPoisoned() && commitPoisonedEvent(ev)) { LOGGER.error( DataNodePipeMessages .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_POISON_MESSAGE_DETECTED_NACKCOUNT_3A9255FB, @@ -3276,6 +3276,15 @@ public class ConsensusPrefetchingQueue { // ======================== Recycle ======================== + private boolean commitPoisonedEvent(final SubscriptionEvent event) { + // Forced ACKs must consume the outstanding mapping before discarding the event. Otherwise, + // later ACKs stay behind a gap that can no longer be redelivered. A failed commit leaves the + // event queued for retry, just as an ordinary ACK keeps its in-flight event on failure. + return commitAndRefreshWalRetention( + extractCommitWriterId(event.getCommitContext()), + extractCommitWriterProgress(event.getCommitContext())); + } + /** Recycles in-flight events that are pollable (timed out) back to the prefetching queue. */ private void recycleInFlightEvents() { for (final InFlightEventKey key : new ArrayList<>(inFlightEvents.keySet())) { @@ -3291,7 +3300,7 @@ public class ConsensusPrefetchingQueue { } if (ev.pollable()) { ev.nack(); - if (ev.isPoisoned()) { + if (ev.isPoisoned() && commitPoisonedEvent(ev)) { LOGGER.error( DataNodePipeMessages .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_POISON_MESSAGE_DETECTED_DURING_23159F02, diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java index 0b9e12eda48..4c44c07d092 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java @@ -98,6 +98,7 @@ import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -3061,6 +3062,177 @@ public class ConsensusPrefetchingQueueTest { } } + @Test + public void testNackPoisonedEventAdvancesCommittedProgress() throws Exception { + assertPoisonedEventProgress(false, false, false); + } + + @Test + public void testSilentNackPoisonedEventAdvancesCommittedProgress() throws Exception { + assertPoisonedEventProgress(true, false, false); + } + + @Test + public void testRecycledPoisonedEventAdvancesCommittedProgress() throws Exception { + assertPoisonedEventProgress(false, true, false); + } + + @Test + public void testPoisonedEventCommitFailureRetainsEvent() throws Exception { + assertPoisonedEventProgress(false, false, true); + } + + private void assertPoisonedEventProgress( + final boolean silent, final boolean recycle, final boolean failCommit) throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final int originalRecycleInterval = + CommonDescriptor.getInstance() + .getConfig() + .getSubscriptionRecycleUncommittedEventIntervalMs(); + final int originalBatchMaxTablets = + CommonDescriptor.getInstance().getConfig().getSubscriptionConsensusBatchMaxTabletCount(); + final int originalBatchMaxDelay = + CommonDescriptor.getInstance().getConfig().getSubscriptionConsensusBatchMaxDelayInMs(); + final File systemDir = temporaryFolder.newFolder("system-poisoned-event-progress"); + ConsensusPrefetchingQueue queue = null; + try { + CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusBatchMaxTabletCount(1); + CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusBatchMaxDelayInMs(0); + if (recycle) { + CommonDescriptor.getInstance() + .getConfig() + .setSubscriptionRecycleUncommittedEventIntervalMs(-1); + } + final DataRegionId regionId = new DataRegionId(1); + final FakeConsensusReqReader reader = new FakeConsensusReqReader(); + final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); + when(serverImpl.getConsensusReqReader()).thenReturn(reader); + when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + + final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); + when(converter.convert(any())) + .thenReturn( + Collections.singletonList(createTablet()), + Collections.emptyList(), + Collections.singletonList(createTablet())); + when(converter.getDatabaseName()).thenReturn("db"); + final ConsensusSubscriptionCommitManager manager = + spy( + newCommitManager( + systemDir, + (consumerGroupId, topicName, id) -> + ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.available( + new RegionProgress(Collections.emptyMap())))); + + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + "topic", + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + regionId, + serverImpl, + new SubscriptionWalRetentionPolicy( + "topic", + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + converter, + manager, + new RegionProgress(Collections.emptyMap()), + 1L, + 1L, + true); + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(1024L * 1024L)); + + // The second request is filtered out, so its automatic commit waits behind the first event. + reader.currentSearchIndex = 2L; + assertTrue(pendingEntries(queue).offer(createRequest(1L))); + assertTrue(pendingEntries(queue).offer(createRequest(2L))); + assertNull(queue.poll("consumer")); + queue.drivePrefetchOnce(); + final SubscriptionEvent event = queue.poll("consumer"); + assertNotNull(event); + final long retainedBytes = queue.getRetainedTabletBytes(); + assertTrue(retainedBytes > 0L); + + if (failCommit) { + final AtomicBoolean rejectFirstCommit = new AtomicBoolean(true); + doAnswer( + invocation -> + rejectFirstCommit.getAndSet(false) ? false : invocation.callRealMethod()) + .when(manager) + .commit( + eq("consumerGroup"), + eq("topic"), + eq(regionId), + eq(event.getCommitContext().getWriterId()), + eq(event.getCommitContext().getWriterProgress())); + } + + for (int retry = 1; retry <= 10; retry++) { + if (recycle) { + queue.drivePrefetchOnce(); + } else if (silent) { + assertTrue(queue.nackSilent("consumer", event.getCommitContext())); + } else { + assertTrue(queue.nack("consumer", event.getCommitContext())); + } + assertEquals(retry, event.getNackCount()); + assertEquals(0L, queue.getSubscriptionUncommittedEventCount()); + if (retry < 10) { + assertFalse(event.isCommitted()); + assertEquals(retainedBytes, queue.getRetainedTabletBytes()); + assertSame(event, queue.poll("consumer")); + } + } + + if (failCommit) { + // A failed forced commit must leave the event and its memory available for a later ACK. + assertFalse(event.isCommitted()); + assertEquals(retainedBytes, queue.getRetainedTabletBytes()); + assertSame(event, queue.poll("consumer")); + assertTrue(queue.ack("consumer", event.getCommitContext())); + } + assertTrue(event.isCommitted()); + assertEquals(0, queue.getPrefetchedEventCount()); + assertEquals(0L, queue.getRetainedTabletBytes()); + assertEquals( + new WriterProgress(1002L, 2L), + manager + .getCommittedRegionProgress("consumerGroup", "topic", regionId) + .getWriterPositions() + .get(event.getCommitContext().getWriterId())); + + // A later data event must remain committable after the poisoned event has been discarded. + reader.currentSearchIndex = 3L; + serverImpl.getWriterSafeFrontierTracker().recordAppliedProgress(1003L, 7, 3L); + assertTrue(pendingEntries(queue).offer(createRequest(3L))); + queue.drivePrefetchOnce(); + final SubscriptionEvent nextEvent = queue.poll("consumer"); + assertNotNull(nextEvent); + assertTrue(queue.ack("consumer", nextEvent.getCommitContext())); + assertEquals( + new WriterProgress(1003L, 3L), + manager + .getCommittedRegionProgress("consumerGroup", "topic", regionId) + .getWriterPositions() + .get(nextEvent.getCommitContext().getWriterId())); + } finally { + if (queue != null) { + queue.close(); + } + CommonDescriptor.getInstance() + .getConfig() + .setSubscriptionRecycleUncommittedEventIntervalMs(originalRecycleInterval); + CommonDescriptor.getInstance() + .getConfig() + .setSubscriptionConsensusBatchMaxTabletCount(originalBatchMaxTablets); + CommonDescriptor.getInstance() + .getConfig() + .setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay); + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + @Test public void testDeactivationReleasesMaterializedTabletMemory() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir();
