From 6216488c3e0465533d341411ff65a7784d680c04 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 6 Nov 2026 10:03:35 +0800 Subject: [PATCH 1/6] Prevent subscription progress rollback on leader activation --- .../ConsensusSubscriptionCommitManager.java | 49 ++++++++++++++++--- .../ConsensusSubscriptionCommitStateTest.java | 33 +++++++++++++ 2 files changed, 75 insertions(+), 7 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java index b183ca745f9bb..dec1c0b9cd95e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java @@ -525,7 +525,7 @@ public boolean refreshPendingInitialProgress( } } - /** Refreshes an exact authoritative frontier for preferred-writer activation. */ + /** Refreshes the activation frontier without regressing newer locally replicated progress. */ public ConfigNodeProgressQueryResult refreshAuthoritativeProgress( final String consumerGroupId, final String topicName, final ConsensusGroupId regionId) { final CommitStateKey stateKey = @@ -545,17 +545,19 @@ public ConfigNodeProgressQueryResult refreshAuthoritativeProgress( if (!queryResult.isAvailable() && !startNewRegionFromBeginning) { return queryResult; } - final RegionProgress authoritativeProgress = - startNewRegionFromBeginning - ? new RegionProgress(Collections.emptyMap()) - : queryResult.getRegionProgress(); final ConsensusSubscriptionCommitState state = commitStates.computeIfAbsent( stateKey.encodedStateKey, ignored -> new ConsensusSubscriptionCommitState( stateKey.regionIdStr, new SubscriptionConsensusProgress())); - state.resetForSeek(authoritativeProgress); + final RegionProgress refreshedProgress = + mergeRegionProgress( + state.getCommittedRegionProgress(), + startNewRegionFromBeginning + ? new RegionProgress(Collections.emptyMap()) + : queryResult.getRegionProgress()); + state.resetForSeek(refreshedProgress); persistRegionProgress(stateKey, state); pendingInitialProgressKeys.remove(stateKey.encodedStateKey); if (!startNewRegionFromBeginning) { @@ -563,7 +565,7 @@ public ConfigNodeProgressQueryResult refreshAuthoritativeProgress( } configNodeProgressAvailableKeys.add(stateKey.encodedStateKey); tailProposalInitializedKeys.add(stateKey.encodedStateKey); - return ConfigNodeProgressQueryResult.available(authoritativeProgress); + return ConfigNodeProgressQueryResult.available(refreshedProgress); } } @@ -1578,6 +1580,39 @@ private static RegionProgress copyRegionProgress(final RegionProgress regionProg : Collections.emptyMap()); } + private static RegionProgress mergeRegionProgress( + final RegionProgress currentProgress, final RegionProgress refreshedProgress) { + final Map mergedWriterPositions = new LinkedHashMap<>(); + mergeWriterPositions(mergedWriterPositions, currentProgress); + mergeWriterPositions(mergedWriterPositions, refreshedProgress); + return new RegionProgress(mergedWriterPositions); + } + + private static void mergeWriterPositions( + final Map mergedWriterPositions, + final RegionProgress regionProgress) { + if (Objects.isNull(regionProgress)) { + return; + } + regionProgress + .getWriterPositions() + .forEach( + (writerId, writerProgress) -> { + if (Objects.isNull(writerId) || Objects.isNull(writerProgress)) { + return; + } + mergedWriterPositions.merge( + writerId, + writerProgress, + (current, refreshed) -> + new ProgressKey(writerId, refreshed) + .compareTo(new ProgressKey(writerId, current)) + > 0 + ? refreshed + : current); + }); + } + @FunctionalInterface interface ConfigNodeProgressFetcher { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java index 9734e19593f1e..68e450fbd9687 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java @@ -330,6 +330,39 @@ public void testNewRegionStartsFromBeginningWhenConfigNodeExplicitlyHasNoProgres } } + @Test + public void testAuthoritativeRefreshPreservesReplicatedProgressWhenConfigNodeIsEmpty() + throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("refreshDoesNotRegressReplicatedProgress"); + try { + final DataRegionId regionId = new DataRegionId(27); + final WriterId writerId = new WriterId(regionId.toString(), 7); + final WriterProgress replicatedProgress = new WriterProgress(200L, 2L); + final ConsensusSubscriptionCommitManager manager = + newCommitManager( + systemDir, + (consumerGroupId, topicName, ignoredRegionId) -> + ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.available( + new RegionProgress(Collections.emptyMap()))); + manager.receiveProgressBroadcast( + "cg", "topic", regionId.toString(), writerId, replicatedProgress); + + final RegionProgress refreshedProgress = + manager.refreshAuthoritativeProgress("cg", "topic", regionId).getRegionProgress(); + + assertEquals(replicatedProgress, refreshedProgress.getWriterPositions().get(writerId)); + assertEquals( + replicatedProgress, + manager + .getCommittedRegionProgress("cg", "topic", regionId) + .getWriterPositions() + .get(writerId)); + } finally { + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + @Test public void testNewRegionWaitsThroughUnavailableConfigNodeThenAcceptsExplicitAbsence() throws Exception { From 6e63028042a5821e45210482609abb9caf1afe77 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 6 Nov 2026 11:07:19 +0800 Subject: [PATCH 2/6] Resolve pending subscription progress before atomic activation refresh --- .../ConsensusSubscriptionCommitManager.java | 22 +- .../ConsensusSubscriptionCommitStateTest.java | 238 ++++++++++++++++++ 2 files changed, 253 insertions(+), 7 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java index dec1c0b9cd95e..5fa8f8b08c1a4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java @@ -551,13 +551,21 @@ public ConfigNodeProgressQueryResult refreshAuthoritativeProgress( ignored -> new ConsensusSubscriptionCommitState( stateKey.regionIdStr, new SubscriptionConsensusProgress())); - final RegionProgress refreshedProgress = - mergeRegionProgress( - state.getCommittedRegionProgress(), - startNewRegionFromBeginning - ? new RegionProgress(Collections.emptyMap()) - : queryResult.getRegionProgress()); - state.resetForSeek(refreshedProgress); + final RegionProgress authoritativeProgress = + startNewRegionFromBeginning + ? new RegionProgress(Collections.emptyMap()) + : queryResult.getRegionProgress(); + final RegionProgress refreshedProgress; + synchronized (state) { + // A pending local-tail proposal is not a committed checkpoint. ConfigNode must resolve it + // exactly before activation can preserve subsequent replicated progress. + refreshedProgress = + pendingInitialProgressKeys.contains(stateKey.encodedStateKey) + ? authoritativeProgress + : mergeRegionProgress(state.getCommittedRegionProgress(), authoritativeProgress); + // Keep the snapshot, merge, and reset atomic with respect to progress broadcasts and ACKs. + state.resetForSeek(refreshedProgress); + } persistRegionProgress(stateKey, state); pendingInitialProgressKeys.remove(stateKey.encodedStateKey); if (!startNewRegionFromBeginning) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java index 68e450fbd9687..7ffe99dcf79cc 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java @@ -35,12 +35,17 @@ import java.io.File; import java.io.FileOutputStream; import java.lang.reflect.Constructor; +import java.lang.reflect.Field; import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.Collections; import java.util.LinkedHashMap; import java.util.Map; +import java.util.concurrent.FutureTask; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import static org.awaitility.Awaitility.await; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotEquals; @@ -363,6 +368,239 @@ public void testAuthoritativeRefreshPreservesReplicatedProgressWhenConfigNodeIsE } } + @Test + public void testAuthoritativeRefreshResolvesPendingTailProposalExactly() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + try { + final DataRegionId regionId = new DataRegionId(28); + final WriterId writerId = new WriterId(regionId.toString(), 7); + for (final RegionProgress configNodeProgress : + Arrays.asList( + new RegionProgress(Collections.emptyMap()), + new RegionProgress( + Collections.singletonMap(writerId, new WriterProgress(100L, 1L))))) { + final File systemDir = temporaryFolder.newFolder(); + final AtomicReference + queryResult = + new AtomicReference<>( + ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.absent()); + final ConsensusSubscriptionCommitManager manager = + newCommitManager(systemDir, (consumerGroupId, topicName, ignored) -> queryResult.get()); + manager.initializeStateFromTailProposal( + "cg", + "topic", + regionId, + new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(200L, 2L)))); + assertTrue(manager.isInitialProgressPending("cg", "topic", regionId)); + + queryResult.set( + ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.available( + configNodeProgress)); + assertEquals( + configNodeProgress, + manager.refreshAuthoritativeProgress("cg", "topic", regionId).getRegionProgress()); + assertEquals( + configNodeProgress, manager.getCommittedRegionProgress("cg", "topic", regionId)); + assertFalse(manager.isInitialProgressPending("cg", "topic", regionId)); + } + } finally { + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + + @Test + public void testAuthoritativeRefreshMergesEachWriterByPhysicalTimeThenLocalSeq() + throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("refreshPerWriterProgress"); + try { + final DataRegionId regionId = new DataRegionId(29); + final WriterId writerA = new WriterId(regionId.toString(), 7); + final WriterId writerB = new WriterId(regionId.toString(), 8); + final WriterId writerC = new WriterId(regionId.toString(), 9); + final WriterId writerD = new WriterId(regionId.toString(), 10); + final WriterId localOnlyWriter = new WriterId(regionId.toString(), 11); + final WriterId remoteOnlyWriter = new WriterId(regionId.toString(), 12); + final Map localPositions = new LinkedHashMap<>(); + localPositions.put(writerA, new WriterProgress(200L, 2L)); + localPositions.put(writerB, new WriterProgress(100L, 99L)); + localPositions.put(writerC, new WriterProgress(200L, 2L)); + localPositions.put(writerD, new WriterProgress(200L, 3L)); + localPositions.put(localOnlyWriter, new WriterProgress(300L, 1L)); + final Map remotePositions = new LinkedHashMap<>(); + remotePositions.put(writerA, new WriterProgress(100L, 999L)); + remotePositions.put(writerB, new WriterProgress(200L, 1L)); + remotePositions.put(writerC, new WriterProgress(200L, 3L)); + remotePositions.put(writerD, new WriterProgress(200L, 2L)); + remotePositions.put(remoteOnlyWriter, new WriterProgress(50L, 1L)); + final ConsensusSubscriptionCommitManager manager = + newCommitManager( + systemDir, + (consumerGroupId, topicName, ignored) -> + ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.available( + new RegionProgress(remotePositions))); + localPositions.forEach( + (writerId, progress) -> + manager.receiveProgressBroadcast( + "cg", "topic", regionId.toString(), writerId, progress)); + final Map expectedPositions = new LinkedHashMap<>(localPositions); + expectedPositions.put(writerB, remotePositions.get(writerB)); + expectedPositions.put(writerC, remotePositions.get(writerC)); + expectedPositions.put(remoteOnlyWriter, remotePositions.get(remoteOnlyWriter)); + final RegionProgress expectedProgress = new RegionProgress(expectedPositions); + + assertEquals( + expectedProgress, + manager.refreshAuthoritativeProgress("cg", "topic", regionId).getRegionProgress()); + assertEquals(expectedProgress, manager.getCommittedRegionProgress("cg", "topic", regionId)); + assertEquals( + expectedProgress, + newCommitManager(systemDir) + .getOrCreateState("cg", "topic", regionId) + .getCommittedRegionProgress()); + } finally { + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + + @Test + public void testAuthoritativeRefreshKeepsExplicitSeekReset() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("refreshAfterSeekReset"); + try { + final DataRegionId regionId = new DataRegionId(30); + final WriterId writerId = new WriterId(regionId.toString(), 7); + final AtomicReference + queryResult = + new AtomicReference<>( + ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.available( + new RegionProgress(Collections.emptyMap()))); + final ConsensusSubscriptionCommitManager manager = + newCommitManager(systemDir, (consumerGroupId, topicName, ignored) -> queryResult.get()); + manager.receiveProgressBroadcast( + "cg", "topic", regionId.toString(), writerId, new WriterProgress(200L, 2L)); + + for (final RegionProgress seekProgress : + Arrays.asList( + new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(100L, 1L))), + new RegionProgress(Collections.emptyMap()))) { + manager.resetState("cg", "topic", regionId, seekProgress); + assertEquals(seekProgress, manager.getCommittedRegionProgress("cg", "topic", regionId)); + queryResult.set( + ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.available( + seekProgress)); + assertEquals( + seekProgress, + manager.refreshAuthoritativeProgress("cg", "topic", regionId).getRegionProgress()); + } + } finally { + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + + @Test + public void testNewRegionActivationPreservesReplicatedProgressWhenConfigNodeHasNoEntry() + throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("newRegionActivationReplicatedProgress"); + try { + final DataRegionId regionId = new DataRegionId(31); + final WriterId writerId = new WriterId(regionId.toString(), 7); + final ConsensusSubscriptionCommitManager manager = + newCommitManager( + systemDir, + (consumerGroupId, topicName, ignored) -> + ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.absent()); + manager.initializeStateWithoutTailProposal("cg", "topic", regionId); + final WriterProgress replicatedProgress = new WriterProgress(200L, 2L); + manager.receiveProgressBroadcast( + "cg", "topic", regionId.toString(), writerId, replicatedProgress); + + assertEquals( + new RegionProgress(Collections.singletonMap(writerId, replicatedProgress)), + manager.refreshAuthoritativeProgress("cg", "topic", regionId).getRegionProgress()); + } finally { + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + + @Test + public void testAuthoritativeRefreshDoesNotOverwriteConcurrentProgressBroadcast() + throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("refreshConcurrentProgressBroadcast"); + Thread broadcastThread = null; + try { + final DataRegionId regionId = new DataRegionId(32); + final WriterId writerId = new WriterId(regionId.toString(), 7); + final ConsensusSubscriptionCommitManager manager = + newCommitManager( + systemDir, + (consumerGroupId, topicName, ignored) -> + ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.available( + new RegionProgress(Collections.emptyMap()))); + manager.getOrCreateState("cg", "topic", regionId); + final AtomicReference afterSnapshot = new AtomicReference<>(); + final ConsensusSubscriptionCommitManager.ConsensusSubscriptionCommitState state = + new ConsensusSubscriptionCommitManager.ConsensusSubscriptionCommitState( + regionId.toString(), new SubscriptionConsensusProgress()) { + @Override + public RegionProgress getCommittedRegionProgress() { + final RegionProgress snapshot = super.getCommittedRegionProgress(); + final Runnable callback = afterSnapshot.getAndSet(null); + if (callback != null) { + callback.run(); + } + return snapshot; + } + }; + final Field statesField = + ConsensusSubscriptionCommitManager.class.getDeclaredField("commitStates"); + statesField.setAccessible(true); + @SuppressWarnings("unchecked") + final Map + states = + (Map) + statesField.get(manager); + states.put(CommitProgressKeeper.generateRegionKey("cg", "topic", regionId.toString()), state); + state.updateFromBroadcast(writerId, new WriterProgress(100L, 1L)); + final WriterProgress latestProgress = new WriterProgress(200L, 2L); + final FutureTask broadcast = + new FutureTask<>( + () -> { + state.updateFromBroadcast(writerId, latestProgress); + return null; + }); + final Thread progressBroadcastThread = new Thread(broadcast); + progressBroadcastThread.setDaemon(true); + broadcastThread = progressBroadcastThread; + afterSnapshot.set( + () -> { + // Deliver a newer broadcast after the activation snapshot was captured. With an + // atomic refresh it waits for the reset; otherwise it is overwritten. + progressBroadcastThread.start(); + await() + .pollInterval(1L, TimeUnit.MILLISECONDS) + .atMost(5L, TimeUnit.SECONDS) + .until( + () -> + broadcast.isDone() + || progressBroadcastThread.getState() == Thread.State.BLOCKED); + }); + + manager.refreshAuthoritativeProgress("cg", "topic", regionId); + broadcast.get(5L, TimeUnit.SECONDS); + + assertEquals( + latestProgress, state.getCommittedRegionProgress().getWriterPositions().get(writerId)); + } finally { + if (broadcastThread != null) { + broadcastThread.join(TimeUnit.SECONDS.toMillis(5L)); + } + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + @Test public void testNewRegionWaitsThroughUnavailableConfigNodeThenAcceptsExplicitAbsence() throws Exception { From fc9e9e9deb1f68adbdc419290d81ada6f1ce60a5 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 6 Nov 2026 11:38:17 +0800 Subject: [PATCH 3/6] Preserve explicit seek progress during activation refresh --- .../ConsensusSubscriptionCommitManager.java | 92 ++++++++++++++----- .../ConsensusSubscriptionCommitStateTest.java | 6 +- 2 files changed, 75 insertions(+), 23 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java index 5fa8f8b08c1a4..9e926c0b49077 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java @@ -160,6 +160,9 @@ public class ConsensusSubscriptionCommitManager { /** New regions that may start from the beginning once ConfigNode explicitly reports no entry. */ private final Set pendingNewRegionProgressKeys = ConcurrentHashMap.newKeySet(); + /** Explicit seeks take precedence over stale, higher progress retained by ConfigNode. */ + private final Set explicitSeekResetKeys = ConcurrentHashMap.newKeySet(); + /** * Regions whose latest initialization/refresh obtained an authoritative ConfigNode resolution. * This includes an explicitly present field and explicit absence for a newly-created region. @@ -280,7 +283,8 @@ SetupSnapshot captureSetupSnapshot(final String consumerGroupId, final Set candidatePositions = + Objects.nonNull(candidateProgress) + ? candidateProgress.getWriterPositions() + : Collections.emptyMap(); + for (final Map.Entry entry : + baselineProgress.getWriterPositions().entrySet()) { + if (Objects.isNull(entry.getKey()) || Objects.isNull(entry.getValue())) { + continue; + } + final WriterProgress candidate = candidatePositions.get(entry.getKey()); + if (Objects.isNull(candidate) + || new ProgressKey(entry.getKey(), candidate) + .compareTo(new ProgressKey(entry.getKey(), entry.getValue())) + < 0) { + return false; + } + } + return true; + } + private static void mergeWriterPositions( final Map mergedWriterPositions, final RegionProgress regionProgress) { @@ -1725,17 +1771,21 @@ private static final class StateSetupSnapshot { private final boolean tailProposalInitialized; + private final boolean explicitSeekReset; + private StateSetupSnapshot( final SubscriptionConsensusProgress progress, final boolean initialProgressPending, final boolean newRegionProgressPending, final boolean configNodeProgressAvailable, - final boolean tailProposalInitialized) { + final boolean tailProposalInitialized, + final boolean explicitSeekReset) { this.progress = progress; this.initialProgressPending = initialProgressPending; this.newRegionProgressPending = newRegionProgressPending; this.configNodeProgressAvailable = configNodeProgressAvailable; this.tailProposalInitialized = tailProposalInitialized; + this.explicitSeekReset = explicitSeekReset; } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java index 7ffe99dcf79cc..0126655cc62d7 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java @@ -470,11 +470,13 @@ public void testAuthoritativeRefreshKeepsExplicitSeekReset() throws Exception { try { final DataRegionId regionId = new DataRegionId(30); final WriterId writerId = new WriterId(regionId.toString(), 7); + final RegionProgress staleConfigNodeProgress = + new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(200L, 2L))); final AtomicReference queryResult = new AtomicReference<>( ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.available( - new RegionProgress(Collections.emptyMap()))); + staleConfigNodeProgress)); final ConsensusSubscriptionCommitManager manager = newCommitManager(systemDir, (consumerGroupId, topicName, ignored) -> queryResult.get()); manager.receiveProgressBroadcast( @@ -488,7 +490,7 @@ public void testAuthoritativeRefreshKeepsExplicitSeekReset() throws Exception { assertEquals(seekProgress, manager.getCommittedRegionProgress("cg", "topic", regionId)); queryResult.set( ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.available( - seekProgress)); + staleConfigNodeProgress)); assertEquals( seekProgress, manager.refreshAuthoritativeProgress("cg", "topic", regionId).getRegionProgress()); From 51aa076fb5c0627da04f002103667c1d9a46bc46 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 6 Nov 2026 12:08:57 +0800 Subject: [PATCH 4/6] Preserve subscription progress during activation seeks --- .../consensus/ConsensusPrefetchingQueue.java | 61 +++++--- .../ConsensusSubscriptionCommitManager.java | 27 +++- .../ConsensusPrefetchingQueueSeekTest.java | 147 ++++++++++++++++++ 3 files changed, 214 insertions(+), 21 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 765202b74f234..2350c1a88703c 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 @@ -602,6 +602,7 @@ private static final class PendingSeekRequest { private final long targetSearchIndex; private final RegionProgress committedRegionProgress; private final String seekReason; + private final boolean explicitSeek; private final boolean previousPrefetchInitialized; private final long previousSeekGeneration; private final long targetSeekGeneration; @@ -613,12 +614,14 @@ private PendingSeekRequest( final long targetSearchIndex, final RegionProgress committedRegionProgress, final String seekReason, + final boolean explicitSeek, final boolean previousPrefetchInitialized, final long previousSeekGeneration, final long targetSeekGeneration) { this.targetSearchIndex = targetSearchIndex; this.committedRegionProgress = committedRegionProgress; this.seekReason = seekReason; + this.explicitSeek = explicitSeek; this.previousPrefetchInitialized = previousPrefetchInitialized; this.previousSeekGeneration = previousSeekGeneration; this.targetSeekGeneration = targetSeekGeneration; @@ -4384,15 +4387,27 @@ public void cleanUp() { * denotes an empty WAL. If old files have been reclaimed, replay may start at a later index. */ public void seekToBeginning() { + seekToBeginning(true); + } + + private void seekToBeginning(final boolean explicitSeek) { seekToResolvedPosition( - FIRST_CONSENSUS_SEARCH_INDEX, new RegionProgress(Collections.emptyMap()), "beginning"); + FIRST_CONSENSUS_SEARCH_INDEX, + new RegionProgress(Collections.emptyMap()), + "beginning", + explicitSeek); } /** * Seeks to the current WAL write position. After this, only newly written data will be consumed. */ public void seekToEnd() { - seekToResolvedPosition(nextSearchIndexAfterCurrent(), computeTailRegionProgress(), "end"); + seekToEnd(true); + } + + private void seekToEnd(final boolean explicitSeek) { + seekToResolvedPosition( + nextSearchIndexAfterCurrent(), computeTailRegionProgress(), "end", explicitSeek); } public void seekToRegionProgress(final RegionProgress regionProgress) { @@ -4422,7 +4437,8 @@ public void seekToRegionProgress(final RegionProgress regionProgress) { seekToResolvedPosition( replayTarget.getStartSearchIndex(), replayTarget.getRecoveryRegionProgress(), - "regionProgress"); + "regionProgress", + true); return; case RETRY_LATER: case LOCATE_MISS: @@ -4438,12 +4454,17 @@ public void seekToRegionProgress(final RegionProgress regionProgress) { } public void seekAfterRegionProgress(final RegionProgress regionProgress) { + seekAfterRegionProgress(regionProgress, true); + } + + private void seekAfterRegionProgress( + final RegionProgress regionProgress, final boolean explicitSeek) { if (!(consensusReqReader instanceof WALNode)) { LOGGER.warn( DataNodePipeMessages .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_SEEKAFTERREGIONPROGRESS_NOT_SUPPORTED_55F36BE8, this); - seekToEnd(); + seekToEnd(explicitSeek); return; } final WALNode walNode = (WALNode) consensusReqReader; @@ -4464,7 +4485,8 @@ public void seekAfterRegionProgress(final RegionProgress regionProgress) { seekToResolvedPosition( replayTarget.getStartSearchIndex(), replayTarget.getRecoveryRegionProgress(), - "regionProgressAfter"); + "regionProgressAfter", + explicitSeek); return; case RETRY_LATER: case LOCATE_MISS: @@ -4482,7 +4504,8 @@ public void seekAfterRegionProgress(final RegionProgress regionProgress) { private synchronized void seekToResolvedPosition( final long targetSearchIndex, final RegionProgress committedRegionProgress, - final String seekReason) { + final String seekReason, + final boolean explicitSeek) { final PendingSeekRequest request; acquireWriteLock(); @@ -4500,6 +4523,7 @@ private synchronized void seekToResolvedPosition( targetSearchIndex, committedRegionProgress, seekReason, + explicitSeek, previousPrefetchInitialized, previousSeekGeneration, targetSeekGeneration); @@ -4650,9 +4674,15 @@ private void applySeekResetUnderWriteLock(final PendingSeekRequest request) { writerStates.clear(); clearRecoveryWriterProgress(); materializedProgressByWriter.clear(); - if (Objects.nonNull(request.committedRegionProgress) - && !request.committedRegionProgress.getWriterPositions().isEmpty()) { - installRecoveryWriterProgress(request.committedRegionProgress); + final RegionProgress committedRegionProgress = + commitManager.resetState( + consumerGroupId, + topicName, + consensusGroupId, + request.committedRegionProgress, + request.explicitSeek); + if (!committedRegionProgress.getWriterPositions().isEmpty()) { + installRecoveryWriterProgress(committedRegionProgress); } // 4. Reset WAL read position. The target has been normalized by replay lookup: if the @@ -4668,10 +4698,7 @@ private void applySeekResetUnderWriteLock(final PendingSeekRequest request) { walGapWaitStartTimeMs = 0L; lastWalGapWaitLogTimeMs = 0L; - // 5. Reset commit state to the writer progress immediately before the first re-delivered - // entry so seek/rebind resumes from the intended frontier. - commitManager.resetState( - consumerGroupId, topicName, consensusGroupId, request.committedRegionProgress); + // 5. Refresh WAL retention from the installed committed frontier. refreshCommittedWalRetentionBoundAndNotify(); LOGGER.info( @@ -4680,9 +4707,7 @@ private void applySeekResetUnderWriteLock(final PendingSeekRequest request) { this, request.seekReason, request.targetSearchIndex, - Objects.nonNull(request.committedRegionProgress) - ? request.committedRegionProgress.getWriterPositions().size() - : 0, + committedRegionProgress.getWriterPositions().size(), seekGeneration.get()); } @@ -5873,9 +5898,9 @@ private boolean tryActivatePendingRuntimeState() { final RegionProgress authoritativeProgress = queryResult.getRegionProgress(); if (prefetchInitialized) { if (authoritativeProgress.getWriterPositions().isEmpty()) { - seekToBeginning(); + seekToBeginning(false); } else { - seekAfterRegionProgress(authoritativeProgress); + seekAfterRegionProgress(authoritativeProgress, false); } } pendingActivationRuntimeState = null; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java index 9e926c0b49077..29a4fad25d4fd 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java @@ -837,6 +837,15 @@ public void resetState( final String topicName, final ConsensusGroupId regionId, final RegionProgress regionProgress) { + resetState(consumerGroupId, topicName, regionId, regionProgress, true); + } + + RegionProgress resetState( + final String consumerGroupId, + final String topicName, + final ConsensusGroupId regionId, + final RegionProgress regionProgress, + final boolean explicitSeek) { final CommitStateKey stateKey = getCommitStateKey(consumerGroupId, topicName, regionId.toString()); synchronized (getTopicPersistLock(stateKey.topicFileKey)) { @@ -848,11 +857,23 @@ public void resetState( consumerGroupId, topicName, regionId); - return; + return copyRegionProgress(regionProgress); + } + final RegionProgress resetProgress; + synchronized (state) { + // Activation applies its seek asynchronously, after the authoritative refresh. Preserve + // committed progress received since that refresh; only a user seek may rewind it. + resetProgress = + explicitSeek + ? copyRegionProgress(regionProgress) + : mergeRegionProgress(state.getCommittedRegionProgress(), regionProgress); + state.resetForSeek(resetProgress); + if (explicitSeek) { + explicitSeekResetKeys.add(stateKey.encodedStateKey); + } } - state.resetForSeek(regionProgress); - explicitSeekResetKeys.add(stateKey.encodedStateKey); persistRegionProgress(stateKey, state); + return resetProgress; } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java index d113e172ca3b9..51f7bf1eae2af 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java @@ -40,6 +40,8 @@ import org.apache.iotdb.rpc.subscription.payload.poll.RegionProgress; import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponseType; import org.apache.iotdb.rpc.subscription.payload.poll.TabletsPayload; +import org.apache.iotdb.rpc.subscription.payload.poll.WriterId; +import org.apache.iotdb.rpc.subscription.payload.poll.WriterProgress; import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; @@ -62,8 +64,10 @@ import java.util.List; import java.util.Set; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; @@ -133,6 +137,149 @@ public void testMultiWriterSeekReplaysAfterResettingOrderedReaders() throws Exce Arrays.asList(createRequest(1L, 1L, 7), createRequest(-1L, 10L, 8)), 2L, 0L, true); } + @Test(timeout = 20_000L) + public void testResolvedSeekDoesNotMaskNewerConfigNodeProgress() throws Exception { + assertActivationProgress(false, false); + } + + @Test(timeout = 20_000L) + public void testActivationSeekPreservesConcurrentProgressBroadcast() throws Exception { + assertActivationProgress(false, true); + } + + @Test(timeout = 20_000L) + public void testExplicitSeekSurvivesActivationQueueReset() throws Exception { + assertActivationProgress(true, false); + } + + private void assertActivationProgress( + final boolean explicitSeek, final boolean broadcastDuringActivation) throws Exception { + final int originalPrefetchThreads = + config.getSubscriptionConsensusPrefetchExecutorMaxThreadNum(); + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder(); + ConsensusSubscriptionPrefetchExecutor executor = null; + ConsensusPrefetchingQueue queue = null; + try { + config.setSubscriptionConsensusPrefetchExecutorMaxThreadNum(1); + IoTDBDescriptor.getInstance().getConfig().setSystemDir(systemDir.getAbsolutePath()); + executor = new ConsensusSubscriptionPrefetchExecutor(); + when(manager.getExecutor()).thenReturn(executor); + when(walNode.getCurrentSearchIndex()).thenReturn(10L); + when(walNode.getLogDirectory()).thenReturn(systemDir); + when(server.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + final DataRegionId regionId = new DataRegionId(1); + final WriterId writerId = new WriterId(regionId.toString(), 7); + final RegionProgress emptyProgress = new RegionProgress(Collections.emptyMap()); + final RegionProgress initialProgress = + broadcastDuringActivation || explicitSeek + ? new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(100L, 1L))) + : emptyProgress; + final AtomicReference configNodeProgress = + new AtomicReference<>(initialProgress); + final ConsensusSubscriptionCommitManager commitManager = + new ConsensusSubscriptionCommitManager( + (consumerGroupId, topicName, ignored) -> + ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.available( + configNodeProgress.get())); + commitManager.getOrCreateState("consumerGroup", "topic", regionId); + final AtomicReference duringReplayLookup = new AtomicReference<>(); + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + "topic", + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + regionId, + server, + new SubscriptionWalRetentionPolicy( + "topic", + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + converter, + commitManager, + initialProgress, + 1L, + 1L, + true) { + @Override + protected ReplayLocateDecision locateReplayStartForRegionProgress( + final RegionProgress progress, final boolean seekAfter) { + final Runnable callback = duringReplayLookup.getAndSet(null); + if (callback != null) { + callback.run(); + } + final long startSearchIndex = + progress.getWriterPositions().values().stream() + .mapToLong(WriterProgress::getLocalSeq) + .max() + .orElse(0L) + + 1L; + return ReplayLocateDecision.found(startSearchIndex, progress, "activation replay"); + } + + @Override + protected ProgressWALIterator createSubscriptionWALIterator( + final long startSearchIndex) { + return new ProgressWALIterator(systemDir, startSearchIndex) { + @Override + boolean hasNext(final ScanBudget scanBudget) { + return false; + } + }; + } + }; + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); + queue.poll("consumer"); + queue.setActive(false); + // A seek while inactive initializes the cursor, so activation must apply a worker seek. + if (broadcastDuringActivation) { + queue.seekAfterRegionProgress(initialProgress); + } else { + queue.seekToBeginning(); + } + final RegionProgress broadcastProgress = + new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(200L, 2L))); + if (broadcastDuringActivation) { + // Advance after the manager's activation refresh, before the worker applies the seek. + duringReplayLookup.set( + () -> + commitManager.receiveProgressBroadcast( + "consumerGroup", + "topic", + regionId.toString(), + writerId, + broadcastProgress.getWriterPositions().get(writerId))); + } + queue.applyRuntimeState( + new ConsensusRegionRuntimeState(2L, 7, true, Collections.singleton(7))); + assertNull(duringReplayLookup.get()); + assertEquals( + explicitSeek + ? emptyProgress + : broadcastDuringActivation ? broadcastProgress : initialProgress, + commitManager.getCommittedRegionProgress("consumerGroup", "topic", regionId)); + + final RegionProgress newerConfigNodeProgress = + new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(300L, 3L))); + configNodeProgress.set(newerConfigNodeProgress); + queue.setActive(false); + queue.applyRuntimeState( + new ConsensusRegionRuntimeState(3L, 7, true, Collections.singleton(7))); + assertEquals( + explicitSeek ? emptyProgress : newerConfigNodeProgress, + commitManager.getCommittedRegionProgress("consumerGroup", "topic", regionId)); + } finally { + if (queue != null) { + queue.close(); + } + if (executor != null) { + executor.shutdown(); + } + config.setSubscriptionConsensusPrefetchExecutorMaxThreadNum(originalPrefetchThreads); + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + private void assertBeginningReplay( final List requests, final long expectedNextIndex, From 27ee0a51f2cc281688edf73c986796578bae54e9 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 6 Nov 2026 12:53:34 +0800 Subject: [PATCH 5/6] Preserve explicit seeks during activation lookup --- .../consensus/ConsensusPrefetchingQueue.java | 55 +++++++++++++------ .../ConsensusPrefetchingQueueSeekTest.java | 30 ++++++++-- 2 files changed, 65 insertions(+), 20 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 2350c1a88703c..c4df0d2b2ce88 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 @@ -4387,27 +4387,32 @@ public void cleanUp() { * denotes an empty WAL. If old files have been reclaimed, replay may start at a later index. */ public void seekToBeginning() { - seekToBeginning(true); + seekToBeginning(true, -1L); } - private void seekToBeginning(final boolean explicitSeek) { + private void seekToBeginning(final boolean explicitSeek, final long expectedSeekGeneration) { seekToResolvedPosition( FIRST_CONSENSUS_SEARCH_INDEX, new RegionProgress(Collections.emptyMap()), "beginning", - explicitSeek); + explicitSeek, + expectedSeekGeneration); } /** * Seeks to the current WAL write position. After this, only newly written data will be consumed. */ public void seekToEnd() { - seekToEnd(true); + seekToEnd(true, -1L); } - private void seekToEnd(final boolean explicitSeek) { + private void seekToEnd(final boolean explicitSeek, final long expectedSeekGeneration) { seekToResolvedPosition( - nextSearchIndexAfterCurrent(), computeTailRegionProgress(), "end", explicitSeek); + nextSearchIndexAfterCurrent(), + computeTailRegionProgress(), + "end", + explicitSeek, + expectedSeekGeneration); } public void seekToRegionProgress(final RegionProgress regionProgress) { @@ -4438,7 +4443,8 @@ public void seekToRegionProgress(final RegionProgress regionProgress) { replayTarget.getStartSearchIndex(), replayTarget.getRecoveryRegionProgress(), "regionProgress", - true); + true, + -1L); return; case RETRY_LATER: case LOCATE_MISS: @@ -4454,17 +4460,19 @@ public void seekToRegionProgress(final RegionProgress regionProgress) { } public void seekAfterRegionProgress(final RegionProgress regionProgress) { - seekAfterRegionProgress(regionProgress, true); + seekAfterRegionProgress(regionProgress, true, -1L); } private void seekAfterRegionProgress( - final RegionProgress regionProgress, final boolean explicitSeek) { + final RegionProgress regionProgress, + final boolean explicitSeek, + final long expectedSeekGeneration) { if (!(consensusReqReader instanceof WALNode)) { LOGGER.warn( DataNodePipeMessages .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_SEEKAFTERREGIONPROGRESS_NOT_SUPPORTED_55F36BE8, this); - seekToEnd(explicitSeek); + seekToEnd(explicitSeek, expectedSeekGeneration); return; } final WALNode walNode = (WALNode) consensusReqReader; @@ -4486,7 +4494,8 @@ private void seekAfterRegionProgress( replayTarget.getStartSearchIndex(), replayTarget.getRecoveryRegionProgress(), "regionProgressAfter", - explicitSeek); + explicitSeek, + expectedSeekGeneration); return; case RETRY_LATER: case LOCATE_MISS: @@ -4505,7 +4514,8 @@ private synchronized void seekToResolvedPosition( final long targetSearchIndex, final RegionProgress committedRegionProgress, final String seekReason, - final boolean explicitSeek) { + final boolean explicitSeek, + final long expectedSeekGeneration) { final PendingSeekRequest request; acquireWriteLock(); @@ -4513,6 +4523,11 @@ private synchronized void seekToResolvedPosition( if (isClosed || closeRequested) { return; } + // A user seek completed after activation captured its frontier. Its cursor and recovery + // progress supersede this older lookup, even if that lookup resolved to a higher checkpoint. + if (!explicitSeek && seekGeneration.get() != expectedSeekGeneration) { + return; + } // Fence old commit contexts immediately. The grouped reset itself is applied later by the // prefetch worker so WAL state and queue state still move under the queue's serial context. final boolean previousPrefetchInitialized = prefetchInitialized; @@ -5889,8 +5904,16 @@ private boolean tryActivatePendingRuntimeState() { if (Objects.isNull(pendingActivationRuntimeState)) { return true; } - final ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult queryResult = - commitManager.refreshAuthoritativeProgress(consumerGroupId, topicName, consensusGroupId); + final long activationSeekGeneration; + final ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult queryResult; + synchronized (this) { + // Wait for an already submitted user seek to finish before capturing the activation + // snapshot. A later user seek is detected again when the resolved lookup is submitted. + activationSeekGeneration = seekGeneration.get(); + queryResult = + commitManager.refreshAuthoritativeProgress( + consumerGroupId, topicName, consensusGroupId); + } if (!queryResult.isAvailable()) { return false; } @@ -5898,9 +5921,9 @@ private boolean tryActivatePendingRuntimeState() { final RegionProgress authoritativeProgress = queryResult.getRegionProgress(); if (prefetchInitialized) { if (authoritativeProgress.getWriterPositions().isEmpty()) { - seekToBeginning(false); + seekToBeginning(false, activationSeekGeneration); } else { - seekAfterRegionProgress(authoritativeProgress, false); + seekAfterRegionProgress(authoritativeProgress, false, activationSeekGeneration); } } pendingActivationRuntimeState = null; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java index 51f7bf1eae2af..66da47a2a389e 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java @@ -152,8 +152,21 @@ public void testExplicitSeekSurvivesActivationQueueReset() throws Exception { assertActivationProgress(true, false); } + @Test(timeout = 20_000L) + public void testExplicitSeekDuringActivationOverridesLocatedFrontier() throws Exception { + assertActivationProgress(false, false, true); + } + private void assertActivationProgress( final boolean explicitSeek, final boolean broadcastDuringActivation) throws Exception { + assertActivationProgress(explicitSeek, broadcastDuringActivation, false); + } + + private void assertActivationProgress( + final boolean explicitSeek, + final boolean broadcastDuringActivation, + final boolean seekDuringActivation) + throws Exception { final int originalPrefetchThreads = config.getSubscriptionConsensusPrefetchExecutorMaxThreadNum(); final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); @@ -172,7 +185,7 @@ private void assertActivationProgress( final WriterId writerId = new WriterId(regionId.toString(), 7); final RegionProgress emptyProgress = new RegionProgress(Collections.emptyMap()); final RegionProgress initialProgress = - broadcastDuringActivation || explicitSeek + broadcastDuringActivation || explicitSeek || seekDuringActivation ? new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(100L, 1L))) : emptyProgress; final AtomicReference configNodeProgress = @@ -232,7 +245,7 @@ boolean hasNext(final ScanBudget scanBudget) { queue.poll("consumer"); queue.setActive(false); // A seek while inactive initializes the cursor, so activation must apply a worker seek. - if (broadcastDuringActivation) { + if (broadcastDuringActivation || seekDuringActivation) { queue.seekAfterRegionProgress(initialProgress); } else { queue.seekToBeginning(); @@ -250,14 +263,23 @@ boolean hasNext(final ScanBudget scanBudget) { writerId, broadcastProgress.getWriterPositions().get(writerId))); } + if (seekDuringActivation) { + // Complete a user seek after activation captured its frontier, before it queues the + // automatic worker reset. The older lookup must not supersede this cursor and progress. + final ConsensusPrefetchingQueue activatedQueue = queue; + duringReplayLookup.set(activatedQueue::seekToBeginning); + } queue.applyRuntimeState( new ConsensusRegionRuntimeState(2L, 7, true, Collections.singleton(7))); assertNull(duringReplayLookup.get()); assertEquals( - explicitSeek + explicitSeek || seekDuringActivation ? emptyProgress : broadcastDuringActivation ? broadcastProgress : initialProgress, commitManager.getCommittedRegionProgress("consumerGroup", "topic", regionId)); + if (seekDuringActivation) { + assertEquals(1L, queue.getCurrentReadSearchIndex()); + } final RegionProgress newerConfigNodeProgress = new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(300L, 3L))); @@ -266,7 +288,7 @@ boolean hasNext(final ScanBudget scanBudget) { queue.applyRuntimeState( new ConsensusRegionRuntimeState(3L, 7, true, Collections.singleton(7))); assertEquals( - explicitSeek ? emptyProgress : newerConfigNodeProgress, + explicitSeek || seekDuringActivation ? emptyProgress : newerConfigNodeProgress, commitManager.getCommittedRegionProgress("consumerGroup", "topic", regionId)); } finally { if (queue != null) { From d16f351791fc5bdff0e6db97297e57db3606a2bd Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 6 Nov 2026 16:26:28 +0800 Subject: [PATCH 6/6] Preserve subscription progress across activation races --- .../consensus/ConsensusPrefetchingQueue.java | 10 +- .../ConsensusSubscriptionCommitManager.java | 255 +++++++++++++----- .../SubscriptionConsensusProgress.java | 29 +- .../ConsensusPrefetchingQueueSeekTest.java | 87 +++++- .../ConsensusSubscriptionCommitStateTest.java | 85 +++++- 5 files changed, 392 insertions(+), 74 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 c4df0d2b2ce88..bd1a270f22fa5 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 @@ -4695,10 +4695,12 @@ private void applySeekResetUnderWriteLock(final PendingSeekRequest request) { topicName, consensusGroupId, request.committedRegionProgress, - request.explicitSeek); - if (!committedRegionProgress.getWriterPositions().isEmpty()) { - installRecoveryWriterProgress(committedRegionProgress); - } + request.explicitSeek, + progress -> { + if (!progress.getWriterPositions().isEmpty()) { + installRecoveryWriterProgress(progress); + } + }); // 4. Reset WAL read position. The target has been normalized by replay lookup: if the // requested progress points before the first retained replayable WAL entry, it is moved to that diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java index 29a4fad25d4fd..fa79c98a76a5f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java @@ -81,6 +81,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; import java.util.function.LongSupplier; /** @@ -160,6 +161,10 @@ public class ConsensusSubscriptionCommitManager { /** New regions that may start from the beginning once ConfigNode explicitly reports no entry. */ private final Set pendingNewRegionProgressKeys = ConcurrentHashMap.newKeySet(); + /** Broadcast progress received while a fresh local-tail proposal awaits ConfigNode resolution. */ + private final Map> pendingInitialBroadcastProgress = + new ConcurrentHashMap<>(); + /** Explicit seeks take precedence over stale, higher progress retained by ConfigNode. */ private final Set explicitSeekResetKeys = ConcurrentHashMap.newKeySet(); @@ -279,12 +284,14 @@ SetupSnapshot captureSetupSnapshot(final String consumerGroupId, final Set copyPendingInitialBroadcastProgress(final String stateKey) { + final Map progress = pendingInitialBroadcastProgress.get(stateKey); + return Objects.nonNull(progress) ? new LinkedHashMap<>(progress) : Collections.emptyMap(); + } + + private void restorePendingInitialBroadcastProgress( + final String stateKey, final Map progress) { + if (Objects.isNull(progress) || progress.isEmpty()) { + pendingInitialBroadcastProgress.remove(stateKey); + } else { + pendingInitialBroadcastProgress.put(stateKey, new ConcurrentHashMap<>(progress)); + } + } + + private RegionProgress takePendingInitialBroadcastProgress(final String stateKey) { + final Map progress = pendingInitialBroadcastProgress.remove(stateKey); + return new RegionProgress( + Objects.nonNull(progress) ? new LinkedHashMap<>(progress) : Collections.emptyMap()); + } + + private void recordPendingInitialBroadcastProgress( + final String stateKey, final WriterId writerId, final WriterProgress writerProgress) { + if (!pendingInitialProgressKeys.contains(stateKey)) { + return; + } + pendingInitialBroadcastProgress + .computeIfAbsent(stateKey, ignored -> new ConcurrentHashMap<>()) + .merge( + writerId, + writerProgress, + (current, candidate) -> + new ProgressKey(writerId, candidate).compareTo(new ProgressKey(writerId, current)) + > 0 + ? candidate + : current); + } + + private void setExplicitSeekReset( + final String stateKey, + final ConsensusSubscriptionCommitState state, + final boolean explicitSeekReset) { + if (explicitSeekReset) { + explicitSeekResetKeys.add(stateKey); + } else { + explicitSeekResetKeys.remove(stateKey); + } + state.getProgress().setExplicitSeekReset(explicitSeekReset); + } + + private RegionProgress resolveInitialProgress( + final String stateKey, + final ConsensusSubscriptionCommitState state, + final RegionProgress authoritativeProgress) { + final RegionProgress pendingBroadcastProgress = takePendingInitialBroadcastProgress(stateKey); + if (explicitSeekResetKeys.contains(stateKey)) { + final RegionProgress refreshedProgress = + mergeRegionProgress(state.getCommittedRegionProgress(), pendingBroadcastProgress); + setExplicitSeekReset( + stateKey, state, !isRegionProgressAtLeast(refreshedProgress, authoritativeProgress)); + return refreshedProgress; + } + setExplicitSeekReset(stateKey, state, false); + return mergeRegionProgress(authoritativeProgress, pendingBroadcastProgress); + } + /** * Initializes an already-bound existing-region queue from its local WAL-tail proposal. * @@ -432,18 +508,25 @@ public void initializeStateFromTailProposal( ignored -> new ConsensusSubscriptionCommitState( stateKey.regionIdStr, new SubscriptionConsensusProgress())); - final ConfigNodeProgressQueryResult queryResult = - configNodeProgressFetcher.fetch(consumerGroupId, topicName, regionId); - if (queryResult.isAvailable()) { - localState.resetForSeek(queryResult.getRegionProgress()); - pendingInitialProgressKeys.remove(stateKey.encodedStateKey); - configNodeProgressAvailableKeys.add(stateKey.encodedStateKey); - } else { - if (!hasLocalPersistedState) { - localState.resetForSeek(copyRegionProgress(tailProposal)); + synchronized (localState) { + if (!pendingInitialProgressKeys.contains(stateKey.encodedStateKey)) { + pendingInitialBroadcastProgress.remove(stateKey.encodedStateKey); } pendingInitialProgressKeys.add(stateKey.encodedStateKey); - configNodeProgressAvailableKeys.remove(stateKey.encodedStateKey); + final ConfigNodeProgressQueryResult queryResult = + configNodeProgressFetcher.fetch(consumerGroupId, topicName, regionId); + if (queryResult.isAvailable()) { + localState.resetForSeek( + resolveInitialProgress( + stateKey.encodedStateKey, localState, queryResult.getRegionProgress())); + pendingInitialProgressKeys.remove(stateKey.encodedStateKey); + configNodeProgressAvailableKeys.add(stateKey.encodedStateKey); + } else { + if (!hasLocalPersistedState) { + localState.resetForSeek(copyRegionProgress(tailProposal)); + } + configNodeProgressAvailableKeys.remove(stateKey.encodedStateKey); + } } persistRegionProgress(stateKey, localState); } @@ -463,28 +546,39 @@ public void initializeStateWithoutTailProposal( ignored -> new ConsensusSubscriptionCommitState( stateKey.regionIdStr, new SubscriptionConsensusProgress())); - final ConfigNodeProgressQueryResult queryResult = - configNodeProgressFetcher.fetch(consumerGroupId, topicName, regionId); - if (queryResult.isAvailable()) { - localState.resetForSeek(queryResult.getRegionProgress()); - pendingInitialProgressKeys.remove(stateKey.encodedStateKey); - pendingNewRegionProgressKeys.remove(stateKey.encodedStateKey); - configNodeProgressAvailableKeys.add(stateKey.encodedStateKey); - persistRegionProgress(stateKey, localState); - } else if (queryResult.getAvailability() == ConfigNodeProgressAvailability.ABSENT - && (!hasLocalPersistedState - || pendingNewRegionProgressKeys.contains(stateKey.encodedStateKey))) { - localState.resetForSeek(new RegionProgress(Collections.emptyMap())); - pendingInitialProgressKeys.remove(stateKey.encodedStateKey); - pendingNewRegionProgressKeys.add(stateKey.encodedStateKey); - configNodeProgressAvailableKeys.add(stateKey.encodedStateKey); - persistRegionProgress(stateKey, localState); - } else { + synchronized (localState) { + if (!pendingInitialProgressKeys.contains(stateKey.encodedStateKey)) { + pendingInitialBroadcastProgress.remove(stateKey.encodedStateKey); + } pendingInitialProgressKeys.add(stateKey.encodedStateKey); - if (!hasLocalPersistedState) { + final ConfigNodeProgressQueryResult queryResult = + configNodeProgressFetcher.fetch(consumerGroupId, topicName, regionId); + if (queryResult.isAvailable()) { + localState.resetForSeek( + resolveInitialProgress( + stateKey.encodedStateKey, localState, queryResult.getRegionProgress())); + pendingInitialProgressKeys.remove(stateKey.encodedStateKey); + pendingNewRegionProgressKeys.remove(stateKey.encodedStateKey); + configNodeProgressAvailableKeys.add(stateKey.encodedStateKey); + persistRegionProgress(stateKey, localState); + } else if (queryResult.getAvailability() == ConfigNodeProgressAvailability.ABSENT + && (!hasLocalPersistedState + || pendingNewRegionProgressKeys.contains(stateKey.encodedStateKey))) { + localState.resetForSeek( + resolveInitialProgress( + stateKey.encodedStateKey, + localState, + new RegionProgress(Collections.emptyMap()))); + pendingInitialProgressKeys.remove(stateKey.encodedStateKey); pendingNewRegionProgressKeys.add(stateKey.encodedStateKey); + configNodeProgressAvailableKeys.add(stateKey.encodedStateKey); + persistRegionProgress(stateKey, localState); + } else { + if (!hasLocalPersistedState) { + pendingNewRegionProgressKeys.add(stateKey.encodedStateKey); + } + configNodeProgressAvailableKeys.remove(stateKey.encodedStateKey); } - configNodeProgressAvailableKeys.remove(stateKey.encodedStateKey); } } } @@ -504,30 +598,34 @@ public boolean refreshPendingInitialProgress( if (!pendingInitialProgressKeys.contains(stateKey.encodedStateKey)) { return true; } - final ConfigNodeProgressQueryResult queryResult = - configNodeProgressFetcher.fetch(consumerGroupId, topicName, regionId); - final boolean startNewRegionFromBeginning = - queryResult.getAvailability() == ConfigNodeProgressAvailability.ABSENT - && pendingNewRegionProgressKeys.contains(stateKey.encodedStateKey); - if (!queryResult.isAvailable() && !startNewRegionFromBeginning) { - return false; - } final ConsensusSubscriptionCommitState state = commitStates.computeIfAbsent( stateKey.encodedStateKey, ignored -> new ConsensusSubscriptionCommitState( stateKey.regionIdStr, new SubscriptionConsensusProgress())); - state.resetForSeek( - startNewRegionFromBeginning - ? new RegionProgress(Collections.emptyMap()) - : queryResult.getRegionProgress()); - persistRegionProgress(stateKey, state); - pendingInitialProgressKeys.remove(stateKey.encodedStateKey); - if (!startNewRegionFromBeginning) { - pendingNewRegionProgressKeys.remove(stateKey.encodedStateKey); + synchronized (state) { + final ConfigNodeProgressQueryResult queryResult = + configNodeProgressFetcher.fetch(consumerGroupId, topicName, regionId); + final boolean startNewRegionFromBeginning = + queryResult.getAvailability() == ConfigNodeProgressAvailability.ABSENT + && pendingNewRegionProgressKeys.contains(stateKey.encodedStateKey); + if (!queryResult.isAvailable() && !startNewRegionFromBeginning) { + return false; + } + final RegionProgress authoritativeProgress = + startNewRegionFromBeginning + ? new RegionProgress(Collections.emptyMap()) + : queryResult.getRegionProgress(); + state.resetForSeek( + resolveInitialProgress(stateKey.encodedStateKey, state, authoritativeProgress)); + pendingInitialProgressKeys.remove(stateKey.encodedStateKey); + if (!startNewRegionFromBeginning) { + pendingNewRegionProgressKeys.remove(stateKey.encodedStateKey); + } + configNodeProgressAvailableKeys.add(stateKey.encodedStateKey); } - configNodeProgressAvailableKeys.add(stateKey.encodedStateKey); + persistRegionProgress(stateKey, state); return true; } } @@ -568,22 +666,31 @@ public ConfigNodeProgressQueryResult refreshAuthoritativeProgress( if (explicitSeekResetKeys.contains(stateKey.encodedStateKey)) { // Explicit seek is an intentional reset; ConfigNode can retain a higher pre-seek // checkpoint until every DataNode has reported the replayed progress. - refreshedProgress = localProgress; - if (isRegionProgressAtLeast(localProgress, authoritativeProgress)) { - explicitSeekResetKeys.remove(stateKey.encodedStateKey); - } + refreshedProgress = + mergeRegionProgress( + localProgress, takePendingInitialBroadcastProgress(stateKey.encodedStateKey)); + setExplicitSeekReset( + stateKey.encodedStateKey, + state, + !isRegionProgressAtLeast(refreshedProgress, authoritativeProgress)); } else if (pendingInitialProgressKeys.contains(stateKey.encodedStateKey)) { // A pending local-tail proposal is not a committed checkpoint. ConfigNode must resolve - // it exactly before activation can preserve subsequent replicated progress. - refreshedProgress = authoritativeProgress; + // it exactly, while broadcasts received during the wait remain committed progress. + refreshedProgress = + mergeRegionProgress( + authoritativeProgress, + takePendingInitialBroadcastProgress(stateKey.encodedStateKey)); } else { refreshedProgress = mergeRegionProgress(localProgress, authoritativeProgress); } // Keep the snapshot, merge, and reset atomic with respect to progress broadcasts and ACKs. state.resetForSeek(refreshedProgress); + if (!explicitSeekResetKeys.contains(stateKey.encodedStateKey)) { + setExplicitSeekReset(stateKey.encodedStateKey, state, false); + } + pendingInitialProgressKeys.remove(stateKey.encodedStateKey); } persistRegionProgress(stateKey, state); - pendingInitialProgressKeys.remove(stateKey.encodedStateKey); if (!startNewRegionFromBeginning) { pendingNewRegionProgressKeys.remove(stateKey.encodedStateKey); } @@ -759,6 +866,7 @@ public void removeState( configNodeProgressAvailableKeys.remove(stateKey.encodedStateKey); tailProposalInitializedKeys.remove(stateKey.encodedStateKey); explicitSeekResetKeys.remove(stateKey.encodedStateKey); + pendingInitialBroadcastProgress.remove(stateKey.encodedStateKey); rewriteTopicProgressFiles(stateKey); } @@ -782,6 +890,7 @@ public void removeAllStatesForTopic(final String consumerGroupId, final String t configNodeProgressAvailableKeys.remove(entry.getKey()); tailProposalInitializedKeys.remove(entry.getKey()); explicitSeekResetKeys.remove(entry.getKey()); + pendingInitialBroadcastProgress.remove(entry.getKey()); keyIterator.remove(); } } @@ -846,6 +955,17 @@ RegionProgress resetState( final ConsensusGroupId regionId, final RegionProgress regionProgress, final boolean explicitSeek) { + return resetState( + consumerGroupId, topicName, regionId, regionProgress, explicitSeek, ignored -> {}); + } + + RegionProgress resetState( + final String consumerGroupId, + final String topicName, + final ConsensusGroupId regionId, + final RegionProgress regionProgress, + final boolean explicitSeek, + final Consumer afterReset) { final CommitStateKey stateKey = getCommitStateKey(consumerGroupId, topicName, regionId.toString()); synchronized (getTopicPersistLock(stateKey.topicFileKey)) { @@ -857,7 +977,9 @@ RegionProgress resetState( consumerGroupId, topicName, regionId); - return copyRegionProgress(regionProgress); + final RegionProgress resetProgress = copyRegionProgress(regionProgress); + afterReset.accept(resetProgress); + return resetProgress; } final RegionProgress resetProgress; synchronized (state) { @@ -869,8 +991,9 @@ RegionProgress resetState( : mergeRegionProgress(state.getCommittedRegionProgress(), regionProgress); state.resetForSeek(resetProgress); if (explicitSeek) { - explicitSeekResetKeys.add(stateKey.encodedStateKey); + setExplicitSeekReset(stateKey.encodedStateKey, state, true); } + afterReset.accept(resetProgress); } persistRegionProgress(stateKey, state); return resetProgress; @@ -1129,7 +1252,10 @@ public void receiveProgressBroadcast( final ConsensusSubscriptionCommitState state = commitStates.get(stateKey.encodedStateKey); if (state != null) { // Update only if broadcast is ahead - state.updateFromBroadcast(writerId, writerProgress); + synchronized (state) { + state.updateFromBroadcast(writerId, writerProgress); + recordPendingInitialBroadcastProgress(stateKey.encodedStateKey, writerId, writerProgress); + } persistProgressIfNeeded(stateKey, state); } else { // Create a new state from the broadcast progress @@ -1290,6 +1416,11 @@ private void recoverTopicStatesIfNeeded( final CommitStateKey recoveredStateKey = getCommitStateKey(consumerGroupId, topicName, entry.getKey()); commitStates.putIfAbsent(recoveredStateKey.encodedStateKey, entry.getValue()); + final ConsensusSubscriptionCommitState recoveredState = + commitStates.get(recoveredStateKey.encodedStateKey); + if (recoveredState.getProgress().isExplicitSeekReset()) { + explicitSeekResetKeys.add(recoveredStateKey.encodedStateKey); + } } topicProgressIndexes.put(topicFileKey, snapshot.indexEntries); } catch (final IOException e) { @@ -1794,19 +1925,23 @@ private static final class StateSetupSnapshot { private final boolean explicitSeekReset; + private final Map pendingInitialBroadcastProgress; + private StateSetupSnapshot( final SubscriptionConsensusProgress progress, final boolean initialProgressPending, final boolean newRegionProgressPending, final boolean configNodeProgressAvailable, final boolean tailProposalInitialized, - final boolean explicitSeekReset) { + final boolean explicitSeekReset, + final Map pendingInitialBroadcastProgress) { this.progress = progress; this.initialProgressPending = initialProgressPending; this.newRegionProgressPending = newRegionProgressPending; this.configNodeProgressAvailable = configNodeProgressAvailable; this.tailProposalInitialized = tailProposalInitialized; this.explicitSeekReset = explicitSeekReset; + this.pendingInitialBroadcastProgress = pendingInitialBroadcastProgress; } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionConsensusProgress.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionConsensusProgress.java index 9448a416873aa..0b6a56389de36 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionConsensusProgress.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionConsensusProgress.java @@ -46,14 +46,24 @@ public class SubscriptionConsensusProgress { private final AtomicLong persistenceThrottleCounter; + private volatile boolean explicitSeekReset; + public SubscriptionConsensusProgress() { this(new RegionProgress(Collections.emptyMap()), 0L); } public SubscriptionConsensusProgress( final RegionProgress committedRegionProgress, final long persistenceThrottleCounter) { + this(committedRegionProgress, persistenceThrottleCounter, false); + } + + public SubscriptionConsensusProgress( + final RegionProgress committedRegionProgress, + final long persistenceThrottleCounter, + final boolean explicitSeekReset) { this.committedRegionProgress = normalize(committedRegionProgress); this.persistenceThrottleCounter = new AtomicLong(persistenceThrottleCounter); + this.explicitSeekReset = explicitSeekReset; } public RegionProgress getCommittedRegionProgress() { @@ -64,6 +74,14 @@ public void setCommittedRegionProgress(final RegionProgress committedRegionProgr this.committedRegionProgress = normalize(committedRegionProgress); } + public boolean isExplicitSeekReset() { + return explicitSeekReset; + } + + public void setExplicitSeekReset(final boolean explicitSeekReset) { + this.explicitSeekReset = explicitSeekReset; + } + public WriterId getCommittedWriterId() { return getDerivedCommittedWriterState().writerId; } @@ -83,12 +101,15 @@ public void incrementPersistenceThrottleCounter() { public void serialize(final DataOutputStream stream) throws IOException { committedRegionProgress.serialize(stream); ReadWriteIOUtils.write(persistenceThrottleCounter.get(), stream); + ReadWriteIOUtils.write(explicitSeekReset, stream); } public static SubscriptionConsensusProgress deserialize(final ByteBuffer buffer) { final RegionProgress committedRegionProgress = RegionProgress.deserialize(buffer); final long persistenceThrottleCounter = ReadWriteIOUtils.readLong(buffer); - return new SubscriptionConsensusProgress(committedRegionProgress, persistenceThrottleCounter); + final boolean explicitSeekReset = buffer.hasRemaining() && ReadWriteIOUtils.readBool(buffer); + return new SubscriptionConsensusProgress( + committedRegionProgress, persistenceThrottleCounter, explicitSeekReset); } @Override @@ -101,12 +122,14 @@ public boolean equals(final Object o) { } final SubscriptionConsensusProgress that = (SubscriptionConsensusProgress) o; return persistenceThrottleCounter.get() == that.persistenceThrottleCounter.get() + && explicitSeekReset == that.explicitSeekReset && Objects.equals(committedRegionProgress, that.committedRegionProgress); } @Override public int hashCode() { - return Objects.hash(committedRegionProgress, persistenceThrottleCounter.get()); + return Objects.hash( + committedRegionProgress, persistenceThrottleCounter.get(), explicitSeekReset); } @Override @@ -116,6 +139,8 @@ public String toString() { + committedRegionProgress + ", persistenceThrottleCounter=" + persistenceThrottleCounter.get() + + ", explicitSeekReset=" + + explicitSeekReset + '}'; } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java index 66da47a2a389e..00f7c921f2a03 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java @@ -21,6 +21,7 @@ import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.consensus.ConsensusGroupId; import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType; import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; @@ -65,8 +66,11 @@ import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; +import static org.awaitility.Awaitility.await; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; @@ -157,9 +161,14 @@ public void testExplicitSeekDuringActivationOverridesLocatedFrontier() throws Ex assertActivationProgress(false, false, true); } + @Test(timeout = 20_000L) + public void testWorkerResetInstallsRecoveryFrontierAtomicallyWithBroadcast() throws Exception { + assertActivationProgress(false, false, false, true); + } + private void assertActivationProgress( final boolean explicitSeek, final boolean broadcastDuringActivation) throws Exception { - assertActivationProgress(explicitSeek, broadcastDuringActivation, false); + assertActivationProgress(explicitSeek, broadcastDuringActivation, false, false); } private void assertActivationProgress( @@ -167,6 +176,15 @@ private void assertActivationProgress( final boolean broadcastDuringActivation, final boolean seekDuringActivation) throws Exception { + assertActivationProgress(explicitSeek, broadcastDuringActivation, seekDuringActivation, false); + } + + private void assertActivationProgress( + final boolean explicitSeek, + final boolean broadcastDuringActivation, + final boolean seekDuringActivation, + final boolean broadcastDuringWorkerReset) + throws Exception { final int originalPrefetchThreads = config.getSubscriptionConsensusPrefetchExecutorMaxThreadNum(); final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); @@ -185,16 +203,64 @@ private void assertActivationProgress( final WriterId writerId = new WriterId(regionId.toString(), 7); final RegionProgress emptyProgress = new RegionProgress(Collections.emptyMap()); final RegionProgress initialProgress = - broadcastDuringActivation || explicitSeek || seekDuringActivation + broadcastDuringActivation + || explicitSeek + || seekDuringActivation + || broadcastDuringWorkerReset ? new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(100L, 1L))) : emptyProgress; final AtomicReference configNodeProgress = new AtomicReference<>(initialProgress); + final RegionProgress broadcastProgress = + new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(200L, 2L))); + final AtomicReference queueReference = new AtomicReference<>(); + final AtomicReference recoveryProgressAtReset = new AtomicReference<>(); + final AtomicReference resetBroadcastThread = new AtomicReference<>(); final ConsensusSubscriptionCommitManager commitManager = new ConsensusSubscriptionCommitManager( (consumerGroupId, topicName, ignored) -> ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.available( - configNodeProgress.get())); + configNodeProgress.get())) { + @Override + RegionProgress resetState( + final String consumerGroupId, + final String topicName, + final ConsensusGroupId regionId, + final RegionProgress regionProgress, + final boolean explicitSeek, + final Consumer afterReset) { + return super.resetState( + consumerGroupId, + topicName, + regionId, + regionProgress, + explicitSeek, + progress -> { + afterReset.accept(progress); + if (!explicitSeek && broadcastDuringWorkerReset) { + recoveryProgressAtReset.set(progress); + assertEquals( + "1", queueReference.get().coreReportMessage().get("recoveryWriterCount")); + final Thread broadcastThread = + new Thread( + () -> + receiveProgressBroadcast( + "consumerGroup", + "topic", + regionId.toString(), + writerId, + broadcastProgress.getWriterPositions().get(writerId))); + broadcastThread.setDaemon(true); + resetBroadcastThread.set(broadcastThread); + broadcastThread.start(); + await() + .pollInterval(1L, TimeUnit.MILLISECONDS) + .atMost(5L, TimeUnit.SECONDS) + .until(() -> broadcastThread.getState() == Thread.State.BLOCKED); + } + }); + } + }; commitManager.getOrCreateState("consumerGroup", "topic", regionId); final AtomicReference duringReplayLookup = new AtomicReference<>(); queue = @@ -241,17 +307,16 @@ boolean hasNext(final ScanBudget scanBudget) { }; } }; + queueReference.set(queue); queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); queue.poll("consumer"); queue.setActive(false); // A seek while inactive initializes the cursor, so activation must apply a worker seek. - if (broadcastDuringActivation || seekDuringActivation) { + if (broadcastDuringActivation || seekDuringActivation || broadcastDuringWorkerReset) { queue.seekAfterRegionProgress(initialProgress); } else { queue.seekToBeginning(); } - final RegionProgress broadcastProgress = - new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(200L, 2L))); if (broadcastDuringActivation) { // Advance after the manager's activation refresh, before the worker applies the seek. duringReplayLookup.set( @@ -272,10 +337,18 @@ boolean hasNext(final ScanBudget scanBudget) { queue.applyRuntimeState( new ConsensusRegionRuntimeState(2L, 7, true, Collections.singleton(7))); assertNull(duringReplayLookup.get()); + if (broadcastDuringWorkerReset) { + final Thread broadcastThread = resetBroadcastThread.get(); + broadcastThread.join(TimeUnit.SECONDS.toMillis(5L)); + assertFalse(broadcastThread.isAlive()); + assertEquals(initialProgress, recoveryProgressAtReset.get()); + } assertEquals( explicitSeek || seekDuringActivation ? emptyProgress - : broadcastDuringActivation ? broadcastProgress : initialProgress, + : broadcastDuringActivation || broadcastDuringWorkerReset + ? broadcastProgress + : initialProgress, commitManager.getCommittedRegionProgress("consumerGroup", "topic", regionId)); if (seekDuringActivation) { assertEquals(1L, queue.getCurrentReadSearchIndex()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java index 0126655cc62d7..052b97717efef 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java @@ -244,6 +244,9 @@ public void testFreshTailProposalWaitsForExplicitConfigNodeProgress() throws Exc assertTrue(manager.hasPersistedState("cg", "topic", regionId)); assertTrue(manager.isInitialProgressPending("cg", "topic", regionId)); assertEquals(tailProposal, manager.getCommittedRegionProgress("cg", "topic", regionId)); + final WriterProgress broadcastProgress = new WriterProgress(300L, 3L); + manager.receiveProgressBroadcast( + "cg", "topic", regionId.toString(), writerId, broadcastProgress); assertFalse(manager.refreshPendingInitialProgress("cg", "topic", regionId)); assertFalse(manager.refreshAuthoritativeProgress("cg", "topic", regionId).isAvailable()); @@ -257,13 +260,54 @@ public void testFreshTailProposalWaitsForExplicitConfigNodeProgress() throws Exc manager .getCommittedRegionProgress("cg", "topic", regionId) .getWriterPositions() - .isEmpty()); + .get(writerId) + .equals(broadcastProgress)); assertTrue(manager.refreshAuthoritativeProgress("cg", "topic", regionId).isAvailable()); } finally { IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); } } + @Test + public void testAuthoritativeRefreshPreservesBroadcastDuringPendingTailProposal() + throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("pendingTailBroadcast"); + try { + final DataRegionId regionId = new DataRegionId(33); + final WriterId writerId = new WriterId(regionId.toString(), 7); + final AtomicReference + queryResult = + new AtomicReference<>( + ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.absent()); + final ConsensusSubscriptionCommitManager manager = + newCommitManager(systemDir, (consumerGroupId, topicName, ignored) -> queryResult.get()); + manager.initializeStateFromTailProposal( + "cg", + "topic", + regionId, + new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(200L, 2L)))); + final WriterProgress broadcastProgress = new WriterProgress(300L, 3L); + manager.receiveProgressBroadcast( + "cg", "topic", regionId.toString(), writerId, broadcastProgress); + + queryResult.set( + ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.available( + new RegionProgress( + Collections.singletonMap(writerId, new WriterProgress(100L, 1L))))); + + assertEquals( + broadcastProgress, + manager + .refreshAuthoritativeProgress("cg", "topic", regionId) + .getRegionProgress() + .getWriterPositions() + .get(writerId)); + } finally { + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + @Test public void testFreshTailProposalUsesAvailableConfigNodeProgressExactly() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); @@ -500,6 +544,45 @@ public void testAuthoritativeRefreshKeepsExplicitSeekReset() throws Exception { } } + @Test + public void testExplicitSeekResetSurvivesManagerRecovery() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("seekResetRecovery"); + try { + final DataRegionId regionId = new DataRegionId(34); + final WriterId writerId = new WriterId(regionId.toString(), 7); + final RegionProgress staleConfigNodeProgress = + new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(200L, 2L))); + final RegionProgress seekProgress = + new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(100L, 1L))); + final ConsensusSubscriptionCommitManager manager = + newCommitManager( + systemDir, + (consumerGroupId, topicName, ignored) -> + ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.available( + staleConfigNodeProgress)); + manager.getOrCreateState("cg", "topic", regionId); + manager.resetState("cg", "topic", regionId, seekProgress); + + final ConsensusSubscriptionCommitManager recoveredManager = + newCommitManager( + systemDir, + (consumerGroupId, topicName, ignored) -> + ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.available( + staleConfigNodeProgress)); + + assertEquals( + seekProgress, + recoveredManager + .refreshAuthoritativeProgress("cg", "topic", regionId) + .getRegionProgress()); + assertEquals( + seekProgress, recoveredManager.getCommittedRegionProgress("cg", "topic", regionId)); + } finally { + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + @Test public void testNewRegionActivationPreservesReplicatedProgressWhenConfigNodeHasNoEntry() throws Exception {