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 765202b74f23..bd1a270f22fa 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,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, -1L); + } + + private void seekToBeginning(final boolean explicitSeek, final long expectedSeekGeneration) { seekToResolvedPosition( - FIRST_CONSENSUS_SEARCH_INDEX, new RegionProgress(Collections.emptyMap()), "beginning"); + FIRST_CONSENSUS_SEARCH_INDEX, + new RegionProgress(Collections.emptyMap()), + "beginning", + explicitSeek, + expectedSeekGeneration); } /** * 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, -1L); + } + + private void seekToEnd(final boolean explicitSeek, final long expectedSeekGeneration) { + seekToResolvedPosition( + nextSearchIndexAfterCurrent(), + computeTailRegionProgress(), + "end", + explicitSeek, + expectedSeekGeneration); } public void seekToRegionProgress(final RegionProgress regionProgress) { @@ -4422,7 +4442,9 @@ public void seekToRegionProgress(final RegionProgress regionProgress) { seekToResolvedPosition( replayTarget.getStartSearchIndex(), replayTarget.getRecoveryRegionProgress(), - "regionProgress"); + "regionProgress", + true, + -1L); return; case RETRY_LATER: case LOCATE_MISS: @@ -4438,12 +4460,19 @@ public void seekToRegionProgress(final RegionProgress regionProgress) { } public void seekAfterRegionProgress(final RegionProgress regionProgress) { + seekAfterRegionProgress(regionProgress, true, -1L); + } + + private void seekAfterRegionProgress( + 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(); + seekToEnd(explicitSeek, expectedSeekGeneration); return; } final WALNode walNode = (WALNode) consensusReqReader; @@ -4464,7 +4493,9 @@ public void seekAfterRegionProgress(final RegionProgress regionProgress) { seekToResolvedPosition( replayTarget.getStartSearchIndex(), replayTarget.getRecoveryRegionProgress(), - "regionProgressAfter"); + "regionProgressAfter", + explicitSeek, + expectedSeekGeneration); return; case RETRY_LATER: case LOCATE_MISS: @@ -4482,7 +4513,9 @@ 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 long expectedSeekGeneration) { final PendingSeekRequest request; acquireWriteLock(); @@ -4490,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; @@ -4500,6 +4538,7 @@ private synchronized void seekToResolvedPosition( targetSearchIndex, committedRegionProgress, seekReason, + explicitSeek, previousPrefetchInitialized, previousSeekGeneration, targetSeekGeneration); @@ -4650,10 +4689,18 @@ 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, + 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 @@ -4668,10 +4715,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 +4724,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()); } @@ -5864,8 +5906,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; } @@ -5873,9 +5923,9 @@ private boolean tryActivatePendingRuntimeState() { final RegionProgress authoritativeProgress = queryResult.getRegionProgress(); if (prefetchInitialized) { if (authoritativeProgress.getWriterPositions().isEmpty()) { - seekToBeginning(); + seekToBeginning(false, activationSeekGeneration); } else { - seekAfterRegionProgress(authoritativeProgress); + seekAfterRegionProgress(authoritativeProgress, false, activationSeekGeneration); } } 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 b183ca745f9b..fa79c98a76a5 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,13 @@ 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(); + /** * Regions whose latest initialization/refresh obtained an authoritative ConfigNode resolution. * This includes an explicitly present field and explicit absence for a newly-created region. @@ -276,11 +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. * @@ -425,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); } @@ -456,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); } } } @@ -497,35 +598,39 @@ 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; } } - /** 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,25 +650,53 @@ 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 authoritativeProgress = + startNewRegionFromBeginning + ? new RegionProgress(Collections.emptyMap()) + : queryResult.getRegionProgress(); + final RegionProgress refreshedProgress; + synchronized (state) { + final RegionProgress localProgress = state.getCommittedRegionProgress(); + 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 = + 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, 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); } configNodeProgressAvailableKeys.add(stateKey.encodedStateKey); tailProposalInitializedKeys.add(stateKey.encodedStateKey); - return ConfigNodeProgressQueryResult.available(authoritativeProgress); + return ConfigNodeProgressQueryResult.available(refreshedProgress); } } @@ -732,6 +865,8 @@ public void removeState( pendingNewRegionProgressKeys.remove(stateKey.encodedStateKey); configNodeProgressAvailableKeys.remove(stateKey.encodedStateKey); tailProposalInitializedKeys.remove(stateKey.encodedStateKey); + explicitSeekResetKeys.remove(stateKey.encodedStateKey); + pendingInitialBroadcastProgress.remove(stateKey.encodedStateKey); rewriteTopicProgressFiles(stateKey); } @@ -754,6 +889,8 @@ public void removeAllStatesForTopic(final String consumerGroupId, final String t pendingNewRegionProgressKeys.remove(entry.getKey()); configNodeProgressAvailableKeys.remove(entry.getKey()); tailProposalInitializedKeys.remove(entry.getKey()); + explicitSeekResetKeys.remove(entry.getKey()); + pendingInitialBroadcastProgress.remove(entry.getKey()); keyIterator.remove(); } } @@ -809,20 +946,58 @@ public void resetState( final String topicName, final ConsensusGroupId regionId, final RegionProgress regionProgress) { - final String key = generateKey(consumerGroupId, topicName, regionId); - final ConsensusSubscriptionCommitState state = commitStates.get(key); - if (state == null) { - LOGGER.warn( - DataNodePipeMessages - .PIPE_LOG_CONSENSUSSUBSCRIPTIONCOMMITMANAGER_CANNOT_RESET_UNKNOWN_C469052F, - consumerGroupId, - topicName, - regionId); - return; + resetState(consumerGroupId, topicName, regionId, regionProgress, true); + } + + RegionProgress resetState( + final String consumerGroupId, + final String topicName, + 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)) { + final ConsensusSubscriptionCommitState state = commitStates.get(stateKey.encodedStateKey); + if (state == null) { + LOGGER.warn( + DataNodePipeMessages + .PIPE_LOG_CONSENSUSSUBSCRIPTIONCOMMITMANAGER_CANNOT_RESET_UNKNOWN_C469052F, + consumerGroupId, + topicName, + regionId); + final RegionProgress resetProgress = copyRegionProgress(regionProgress); + afterReset.accept(resetProgress); + return resetProgress; + } + 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) { + setExplicitSeekReset(stateKey.encodedStateKey, state, true); + } + afterReset.accept(resetProgress); + } + persistRegionProgress(stateKey, state); + return resetProgress; } - state.resetForSeek(regionProgress); - persistRegionProgress( - getCommitStateKey(consumerGroupId, topicName, regionId.toString()), state); } /** Persists all states. Should be called during graceful shutdown. */ @@ -1077,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 @@ -1238,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) { @@ -1578,6 +1761,64 @@ 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 boolean isRegionProgressAtLeast( + final RegionProgress candidateProgress, final RegionProgress baselineProgress) { + if (Objects.isNull(baselineProgress)) { + return true; + } + final Map 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) { + 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 { @@ -1682,17 +1923,25 @@ private static final class StateSetupSnapshot { private final boolean tailProposalInitialized; + 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 tailProposalInitialized, + 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 9448a416873a..0b6a56389de3 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 d113e172ca3b..00f7c921f2a0 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; @@ -40,6 +41,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 +65,13 @@ import java.util.List; 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; import static org.mockito.Mockito.times; @@ -133,6 +141,240 @@ 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); + } + + @Test(timeout = 20_000L) + public void testExplicitSeekDuringActivationOverridesLocatedFrontier() throws Exception { + 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, false); + } + + private void assertActivationProgress( + final boolean explicitSeek, + 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(); + 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 + || 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())) { + @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 = + 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; + } + }; + } + }; + 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 || broadcastDuringWorkerReset) { + queue.seekAfterRegionProgress(initialProgress); + } else { + queue.seekToBeginning(); + } + 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))); + } + 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()); + 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 || broadcastDuringWorkerReset + ? 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))); + configNodeProgress.set(newerConfigNodeProgress); + queue.setActive(false); + queue.applyRuntimeState( + new ConsensusRegionRuntimeState(3L, 7, true, Collections.singleton(7))); + assertEquals( + explicitSeek || seekDuringActivation ? 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, 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 9734e19593f1..052b97717efe 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; @@ -239,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()); @@ -252,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(); @@ -330,6 +379,313 @@ 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 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 RegionProgress staleConfigNodeProgress = + new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(200L, 2L))); + final AtomicReference + queryResult = + new AtomicReference<>( + ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.available( + staleConfigNodeProgress)); + 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( + staleConfigNodeProgress)); + assertEquals( + seekProgress, + manager.refreshAuthoritativeProgress("cg", "topic", regionId).getRegionProgress()); + } + } finally { + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + + @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 { + 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 {