This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch revert-silent in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 0901540a3e0d27a5aa2048b5358a332f9c74bb6b Author: Caideyipi <[email protected]> AuthorDate: Wed Sep 30 21:42:20 2026 +0800 Revert "[Fix] Prevent silent subscription data loss on WAL replay gaps (#18741)" This reverts commit 0c9fb83974fd00470506bb82a4563c127e195cde. --- .../apache/iotdb/db/i18n/DataNodePipeMessages.java | 11 +- .../apache/iotdb/db/i18n/DataNodePipeMessages.java | 9 +- .../consensus/ConsensusPrefetchingQueue.java | 110 ++----------- .../consensus/ConsensusPrefetchingQueueTest.java | 172 --------------------- 4 files changed, 24 insertions(+), 278 deletions(-) diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 95583ded3a9..1c0362b627b 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -2216,13 +2216,10 @@ public final class DataNodePipeMessages { "ProgressWALIterator: skipped {} unreadable retained WAL files in directory {}, " + "firstFile={}, lastFile={}, firstError={}; historical subscription data in these " + "files cannot be replayed"; - public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_FOUND_UNAVAILABLE_SEARCH_INDEXES_E0CBFFFA = - "ConsensusPrefetchingQueue {}: WAL replay found unavailable search indexes [{}, {}) before " - + "searchIndex {}; forcing WAL refresh before retry"; - public static final String MESSAGE_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_CANNOT_RECOVER_SEARCH_INDEXES_70781B22 = - "ConsensusPrefetchingQueue %s: WAL replay cannot recover search indexes [%s, %s), " - + "unavailableEntries=%s, totalWalGapSkippedEntries=%s; subscription delivery is " - + "stalled to prevent silent data loss"; + public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64 = + "ConsensusPrefetchingQueue {}: WAL replay skipped unavailable search indexes [{}, {}), " + + "skippedEntries={}, totalWalGapSkippedEntries={}; the missing WAL data may have been " + + "reclaimed before subscription consumption"; public static final String PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28 = "Pipe {}@{}: terminate event committed for historical transfer. creationTime: {}, " + "shouldMark: {}. {}"; diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index ccb9cd42da2..49096659ade 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -2057,12 +2057,9 @@ public final class DataNodePipeMessages { public static final String PIPE_LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_FFC8455E = "ProgressWALIterator：跳过了 {} 个无法读取的保留 WAL 文件，directory={}，firstFile={}，" + "lastFile={}，firstError={}；这些文件中的历史订阅数据无法重放"; - public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_FOUND_UNAVAILABLE_SEARCH_INDEXES_E0CBFFFA = - "ConsensusPrefetchingQueue {}：WAL 回放发现不可用 searchIndex 区间 [{}, {})，下一个可见 " - + "searchIndex={}；正在强制刷新 WAL 后重试"; - public static final String MESSAGE_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_CANNOT_RECOVER_SEARCH_INDEXES_70781B22 = - "ConsensusPrefetchingQueue %s：WAL 回放无法恢复 searchIndex 区间 [%s, %s)，不可用条目数=%s，" - + "walGapSkippedEntries 总数=%s；为防止静默丢数，订阅投递已停滞"; + public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64 = + "ConsensusPrefetchingQueue {}：WAL 重放跳过了不可用的 searchIndex 区间 [{}, {})，" + + "skippedEntries={}，totalWalGapSkippedEntries={}；缺失的 WAL 数据可能已在订阅消费前被回收"; public static final String PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28 = "Pipe {}@{}：历史传输的终止事件已提交。creationTime：{}，shouldMark：{}。{}"; public static final String PIPE_LOG_PIPE_HISTORICAL_SOURCE_HAS_SUPPLIED_ALL_EVENTS_EMITTING_8B58DE19 = 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 c5641db1f18..0dcddbef7a2 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 @@ -297,12 +297,6 @@ public class ConsensusPrefetchingQueue { private volatile long lastWalGapWaitLogTimeMs = 0L; - /** Search index whose WAL visibility gap is being retried after forcing a WAL refresh. */ - private volatile long walGapRetryExpectedSearchIndex = Long.MIN_VALUE; - - /** Critical replay failure returned to consumers instead of silently advancing past data. */ - private volatile String walReplayFailureMessage; - /** Fallback committed region progress from local persisted state. */ private final RegionProgress fallbackCommittedRegionProgress; @@ -777,9 +771,6 @@ public class ConsensusPrefetchingQueue { if (pendingSeekRequest != null) { return null; } - if (Objects.nonNull(walReplayFailureMessage)) { - return generateCriticalErrorResponse(walReplayFailureMessage); - } final SubscriptionEvent event = pollInternal(consumerId); if (Objects.nonNull(event) && prefetchingQueue.size() < MAX_PREFETCHING_QUEUE_SIZE) { requestPrefetch(); @@ -846,7 +837,6 @@ public class ConsensusPrefetchingQueue { // readers without WAL support. this.subscriptionWALIterator = createSubscriptionWALIterator(resolvedStart.getStartSearchIndex()); - resetWalReplayFailureState(); this.prefetchInitialized = true; this.observedSeekGeneration = seekGeneration.get(); discardBatch(this.lingerBatch); @@ -1437,11 +1427,6 @@ public class ConsensusPrefetchingQueue { applyPendingSubscriptionWalReset(observedSeekGeneration); recycleInFlightEvents(); - if (Objects.nonNull(walReplayFailureMessage)) { - blockRealtimeAdmission(); - return PrefetchRoundResult.dormant(); - } - if (!isActive) { blockRealtimeAdmission(); return computeIdleRoundResult(); @@ -1526,11 +1511,6 @@ public class ConsensusPrefetchingQueue { if (batch.isEmpty() && lingerBatch.isEmpty()) { final MaterializationResult walResult = tryCatchUpFromWAL(observedSeekGeneration); - if (walResult == MaterializationResult.WAL_GAP) { - return Objects.nonNull(walReplayFailureMessage) - ? PrefetchRoundResult.dormant() - : PrefetchRoundResult.rescheduleAfter(WAL_GAP_RETRY_SLEEP_MS); - } if (walResult == MaterializationResult.MEMORY_BLOCKED) { blockRealtimeAdmission(); return PrefetchRoundResult.rescheduleAfter(MEMORY_RETRY_SLEEP_MS); @@ -1946,9 +1926,7 @@ public class ConsensusPrefetchingQueue { pumpFromSubscriptionWAL( batchState, expectedSeekGeneration, maxWalEntries, maxTablets, maxBatchBytes); if (materializationResult != MaterializationResult.SUCCESS) { - if ((materializationResult == MaterializationResult.MEMORY_BLOCKED - || materializationResult == MaterializationResult.WAL_GAP) - && !batchState.isEmpty()) { + if (materializationResult == MaterializationResult.MEMORY_BLOCKED && !batchState.isEmpty()) { if (!flushBatch(batchState, expectedSeekGeneration)) { discardBatch(batchState); return MaterializationResult.STALE; @@ -1991,10 +1969,6 @@ public class ConsensusPrefetchingQueue { if (isBeforeLocalCursor(walEntry)) { continue; } - final MaterializationResult continuityResult = validateWalReplayContinuity(walEntry); - if (continuityResult != MaterializationResult.SUCCESS) { - return continuityResult; - } if (shouldSkipForRecoveryProgress(walEntry)) { advanceWalReplayCursorIfPresent(walEntry); continue; @@ -2057,67 +2031,22 @@ public class ConsensusPrefetchingQueue { if (!hasLocalSearchIndex(request)) { return; } - nextExpectedSearchIndex.set(request.getSearchIndex() + 1); - } - - private MaterializationResult validateWalReplayContinuity(final IndexedConsensusRequest request) { - if (!hasLocalSearchIndex(request)) { - return MaterializationResult.SUCCESS; - } final long actualSearchIndex = request.getSearchIndex(); final long expectedSearchIndex = nextExpectedSearchIndex.get(); - if (actualSearchIndex <= expectedSearchIndex) { - if (actualSearchIndex == expectedSearchIndex - && walGapRetryExpectedSearchIndex == expectedSearchIndex) { - walGapRetryExpectedSearchIndex = Long.MIN_VALUE; - pendingWalGapRetryRequested = false; - } - return MaterializationResult.SUCCESS; - } - - if (walGapRetryExpectedSearchIndex != expectedSearchIndex) { - walGapRetryExpectedSearchIndex = expectedSearchIndex; + if (actualSearchIndex > expectedSearchIndex) { + final long skippedEntries = actualSearchIndex - expectedSearchIndex; + final long totalSkippedEntries = walGapSkippedEntries.addAndGet(skippedEntries); LOGGER.warn( DataNodePipeMessages - .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_FOUND_UNAVAILABLE_SEARCH_INDEXES_E0CBFFFA, + .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64, this, expectedSearchIndex, actualSearchIndex, - actualSearchIndex); - if (consensusReqReader instanceof WALNode) { - ((WALNode) consensusReqReader).rollWALFile(); - } - resetSubscriptionWALPosition(expectedSearchIndex); - onWalGapRetryScheduled(); - pendingWalGapRetryRequested = true; - return MaterializationResult.WAL_GAP; - } - - if (Objects.isNull(walReplayFailureMessage)) { - final long unavailableEntries = actualSearchIndex - expectedSearchIndex; - final long totalUnavailableEntries = walGapSkippedEntries.addAndGet(unavailableEntries); - walReplayFailureMessage = - String.format( - DataNodePipeMessages - .MESSAGE_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_CANNOT_RECOVER_SEARCH_INDEXES_70781B22, - this, - expectedSearchIndex, - actualSearchIndex, - unavailableEntries, - totalUnavailableEntries); - LOGGER.error(walReplayFailureMessage); - blockRealtimeAdmission(); + skippedEntries, + totalSkippedEntries); } - return MaterializationResult.WAL_GAP; - } - - private void resetWalReplayFailureState() { - pendingWalGapRetryRequested = false; - walGapWaitStartTimeMs = 0L; - lastWalGapWaitLogTimeMs = 0L; - walGapRetryExpectedSearchIndex = Long.MIN_VALUE; - walReplayFailureMessage = null; + nextExpectedSearchIndex.set(actualSearchIndex + 1); } private void ensureSubscriptionWalReadable() { @@ -3120,7 +3049,9 @@ public class ConsensusPrefetchingQueue { reconcileRetainedTabletMemoryAfterCleanup(); memoryBlockedEntryBytes = -1L; resetBatchWriterProgress(); - resetWalReplayFailureState(); + pendingWalGapRetryRequested = false; + walGapWaitStartTimeMs = 0L; + lastWalGapWaitLogTimeMs = 0L; pendingSubscriptionWalResetSearchIndex = Long.MIN_VALUE; pendingSubscriptionWalResetGeneration = Long.MIN_VALUE; closeSubscriptionWALIterator(); @@ -3414,7 +3345,9 @@ public class ConsensusPrefetchingQueue { reconcileRetainedTabletMemoryAfterCleanup(); resetBatchWriterProgress(); observedSeekGeneration = seekGeneration.get(); - resetWalReplayFailureState(); + pendingWalGapRetryRequested = false; + 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. @@ -3515,7 +3448,6 @@ public class ConsensusPrefetchingQueue { return 0L; } - WALFileUtils.ascSortByVersionId(walFiles); final int replayFileIndex = Math.max(0, WALFileUtils.binarySearchFileBySearchIndex(walFiles, nextExpectedSearchIndex)); return WALFileUtils.parseVersionId(walFiles[replayFileIndex].getName()); @@ -4028,13 +3960,6 @@ public class ConsensusPrefetchingQueue { createNonCommittableContext(IoTDBDescriptor.getInstance().getConfig().getDataNodeId())); } - private SubscriptionEvent generateCriticalErrorResponse(final String errorMessage) { - return new SubscriptionEvent( - SubscriptionPollResponseType.ERROR.getType(), - new ErrorPayload(errorMessage, true), - createNonCommittableContext(IoTDBDescriptor.getInstance().getConfig().getDataNodeId())); - } - private SubscriptionEvent generateOutdatedErrorResponse() { return new SubscriptionEvent( SubscriptionPollResponseType.ERROR.getType(), @@ -4139,7 +4064,9 @@ public class ConsensusPrefetchingQueue { memoryBlockedEntryBytes = -1L; prefetchInitialized = false; observedSeekGeneration = seekGeneration.get(); - resetWalReplayFailureState(); + pendingWalGapRetryRequested = false; + walGapWaitStartTimeMs = 0L; + lastWalGapWaitLogTimeMs = 0L; pendingSubscriptionWalResetSearchIndex = Long.MIN_VALUE; pendingSubscriptionWalResetGeneration = Long.MIN_VALUE; closeSubscriptionWALIterator(); @@ -4227,9 +4154,6 @@ public class ConsensusPrefetchingQueue { if (!isActive) { return SubscriptionProgressSnapshot.STATUS_INACTIVE; } - if (Objects.nonNull(walReplayFailureMessage)) { - return SubscriptionProgressSnapshot.STATUS_STALLED; - } if (getLag() <= 0L) { return SubscriptionProgressSnapshot.STATUS_CAUGHT_UP; } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java index a854b8ad6dc..6efb3298bf9 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java @@ -1382,92 +1382,6 @@ public class ConsensusPrefetchingQueueTest { } } - @Test - public void testWalReplayRetriesGapWithoutSkippingEntries() throws Exception { - final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); - final File systemDir = temporaryFolder.newFolder("wal-replay-gap-counter"); - ConsensusPrefetchingQueue queue = null; - try { - final DataRegionId regionId = new DataRegionId(9); - final WALNode walNode = mock(WALNode.class); - when(walNode.getCurrentSearchIndex()).thenReturn(4L); - when(walNode.getLogDirectory()).thenReturn(systemDir); - final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); - when(serverImpl.getConsensusReqReader()).thenReturn(walNode); - when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); - - final AtomicInteger conversionCount = new AtomicInteger(); - final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); - when(converter.convert(any())) - .thenAnswer( - ignored -> { - conversionCount.incrementAndGet(); - return Collections.singletonList(createTablet()); - }); - when(converter.getDatabaseName()).thenReturn("db"); - - final Iterator<IndexedConsensusRequest> initiallyVisibleEntries = - Arrays.asList(createRequest(1L), createRequest(4L)).iterator(); - final ProgressWALIterator initialIterator = mock(ProgressWALIterator.class); - when(initialIterator.hasNext()).thenAnswer(ignored -> initiallyVisibleEntries.hasNext()); - when(initialIterator.next()).thenAnswer(ignored -> initiallyVisibleEntries.next()); - final Iterator<IndexedConsensusRequest> refreshedEntries = - Arrays.asList(createRequest(2L), createRequest(3L), createRequest(4L)).iterator(); - final ProgressWALIterator refreshedIterator = mock(ProgressWALIterator.class); - when(refreshedIterator.hasNext()).thenAnswer(ignored -> refreshedEntries.hasNext()); - when(refreshedIterator.next()).thenAnswer(ignored -> refreshedEntries.next()); - final AtomicInteger iteratorCreationCount = new AtomicInteger(); - - queue = - new ConsensusPrefetchingQueue( - "consumerGroup", - "topic", - TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, - regionId, - serverImpl, - new SubscriptionWalRetentionPolicy( - "topic", - SubscriptionWalRetentionPolicy.UNBOUNDED, - SubscriptionWalRetentionPolicy.UNBOUNDED), - converter, - newCommitManager(systemDir), - new RegionProgress(Collections.emptyMap()), - 1L, - 1L, - true) { - @Override - protected ProgressWALIterator createSubscriptionWALIterator( - final long startSearchIndex) { - return iteratorCreationCount.getAndIncrement() == 0 - ? initialIterator - : refreshedIterator; - } - }; - queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); - - assertNull(queue.poll("consumer")); - queue.drivePrefetchOnce(); - - assertEquals(1L, queue.getWalPathAcceptedEntries()); - assertEquals(1, conversionCount.get()); - assertEquals(0L, queue.getWalGapSkippedEntries()); - assertEquals(2L, queue.getCurrentReadSearchIndex()); - verify(walNode).rollWALFile(); - - queue.drivePrefetchOnce(); - - assertEquals(4L, queue.getWalPathAcceptedEntries()); - assertEquals(4, conversionCount.get()); - assertEquals(0L, queue.getWalGapSkippedEntries()); - assertEquals(5L, queue.getCurrentReadSearchIndex()); - } finally { - if (queue != null) { - queue.close(); - } - IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); - } - } - @Test public void testPendingGapReplayHonorsPerRoundWalEntryLimit() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); @@ -1609,92 +1523,6 @@ public class ConsensusPrefetchingQueueTest { } } - @Test - public void testWalReplayFailsCriticallyWhenGapRemainsUnavailable() throws Exception { - final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); - final File systemDir = temporaryFolder.newFolder("wal-replay-unrecoverable-gap"); - ConsensusPrefetchingQueue queue = null; - try { - final DataRegionId regionId = new DataRegionId(10); - final FakeConsensusReqReader reader = new FakeConsensusReqReader(); - reader.currentSearchIndex = 4L; - final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); - when(serverImpl.getConsensusReqReader()).thenReturn(reader); - when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); - - final AtomicInteger conversionCount = new AtomicInteger(); - final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); - when(converter.convert(any())) - .thenAnswer( - ignored -> { - conversionCount.incrementAndGet(); - return Collections.singletonList(createTablet()); - }); - when(converter.getDatabaseName()).thenReturn("db"); - - final Iterator<IndexedConsensusRequest> initiallyVisibleEntries = - Arrays.asList(createRequest(1L), createRequest(4L)).iterator(); - final ProgressWALIterator initialIterator = mock(ProgressWALIterator.class); - when(initialIterator.hasNext()).thenAnswer(ignored -> initiallyVisibleEntries.hasNext()); - when(initialIterator.next()).thenAnswer(ignored -> initiallyVisibleEntries.next()); - final Iterator<IndexedConsensusRequest> refreshedEntries = - Collections.singletonList(createRequest(4L)).iterator(); - final ProgressWALIterator refreshedIterator = mock(ProgressWALIterator.class); - when(refreshedIterator.hasNext()).thenAnswer(ignored -> refreshedEntries.hasNext()); - when(refreshedIterator.next()).thenAnswer(ignored -> refreshedEntries.next()); - final AtomicInteger iteratorCreationCount = new AtomicInteger(); - - queue = - new ConsensusPrefetchingQueue( - "consumerGroup", - "topic", - TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, - regionId, - serverImpl, - new SubscriptionWalRetentionPolicy( - "topic", - SubscriptionWalRetentionPolicy.UNBOUNDED, - SubscriptionWalRetentionPolicy.UNBOUNDED), - converter, - newCommitManager(systemDir), - new RegionProgress(Collections.emptyMap()), - 1L, - 1L, - true) { - @Override - protected ProgressWALIterator createSubscriptionWALIterator( - final long startSearchIndex) { - return iteratorCreationCount.getAndIncrement() == 0 - ? initialIterator - : refreshedIterator; - } - }; - queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); - - assertNull(queue.poll("consumer")); - queue.drivePrefetchOnce(); - queue.drivePrefetchOnce(); - - assertEquals(1L, queue.getWalPathAcceptedEntries()); - assertEquals(1, conversionCount.get()); - assertEquals(2L, queue.getWalGapSkippedEntries()); - assertEquals(2L, queue.getCurrentReadSearchIndex()); - assertEquals(4L, queue.getProgressStatus()); - - final SubscriptionEvent errorEvent = queue.poll("consumer"); - assertNotNull(errorEvent); - assertEquals( - SubscriptionPollResponseType.ERROR.getType(), - errorEvent.getCurrentResponse().getResponseType()); - assertTrue(((ErrorPayload) errorEvent.getCurrentResponse().getPayload()).isCritical()); - } finally { - if (queue != null) { - queue.close(); - } - IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); - } - } - @Test public void testActivationRetriesUntilConfigNodeProgressIsExplicitlyAvailable() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir();
