This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch fix/consensus-subscription-read-frontiers in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 92cde8c1571f5bce7485d43a1bdb693d10e00517 Author: Caideyipi <[email protected]> AuthorDate: Tue Oct 13 13:39:12 2026 +0800 [Subscription] Preserve writer order and back off on blocked safe frontiers --- .../consensus/iot/WriterSafeFrontierTracker.java | 33 ++ .../iot/WriterSafeFrontierTrackerTest.java | 20 + .../consensus/ConsensusPrefetchingQueue.java | 378 +++++++++++++++- .../broker/consensus/ProgressWALIterator.java | 31 ++ ...onsensusPrefetchingQueueDataNodeMemoryTest.java | 20 +- .../ConsensusPrefetchingQueueSeekTest.java | 84 ++-- ...onsensusPrefetchingQueueWriterOrderingTest.java | 484 +++++++++++++++++++++ 7 files changed, 1008 insertions(+), 42 deletions(-) diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/WriterSafeFrontierTracker.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/WriterSafeFrontierTracker.java index a8e96e06cee..97e35ef1c3b 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/WriterSafeFrontierTracker.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/WriterSafeFrontierTracker.java @@ -107,6 +107,39 @@ public class WriterSafeFrontierTracker { return Collections.unmodifiableMap(snapshot); } + /** + * Returns each safe time together with the applied sequence it covers. A subscription may use the + * safe time only after consuming that sequence; applying records to the state machine does not + * imply that an independent WAL reader has consumed them. + */ + public synchronized Map<Integer, SafeFrontier> snapshotSafeFrontiers() { + final Map<Integer, SafeFrontier> snapshot = new HashMap<>(); + for (final Map.Entry<Integer, WriterFrontierState> entry : states.entrySet()) { + snapshot.put( + entry.getKey(), + new SafeFrontier(entry.getValue().effectiveSafePt, entry.getValue().appliedLocalSeq)); + } + return Collections.unmodifiableMap(snapshot); + } + + public static final class SafeFrontier { + private final long physicalTime; + private final long localSeq; + + private SafeFrontier(final long physicalTime, final long localSeq) { + this.physicalTime = physicalTime; + this.localSeq = localSeq; + } + + public long getPhysicalTime() { + return physicalTime; + } + + public long getLocalSeq() { + return localSeq; + } + } + private void promotePendingIfReady(final WriterFrontierState state) { if (!state.hasPendingSafeTime()) { return; diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/WriterSafeFrontierTrackerTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/WriterSafeFrontierTrackerTest.java index ce01ceacccc..d18aff9b95a 100644 --- a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/WriterSafeFrontierTrackerTest.java +++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/WriterSafeFrontierTrackerTest.java @@ -21,10 +21,30 @@ package org.apache.iotdb.consensus.iot; import org.junit.Test; +import java.util.Map; + import static org.junit.Assert.assertEquals; public class WriterSafeFrontierTrackerTest { + @Test + public void testSafeFrontierSnapshotCarriesAppliedSequenceAndIsDetached() { + final WriterSafeFrontierTracker tracker = new WriterSafeFrontierTracker(); + tracker.recordAppliedProgress(100L, 7, 10L); + tracker.observePendingWriterSafeTimeBarrier(130L, 7, 20L); + final Map<Integer, WriterSafeFrontierTracker.SafeFrontier> before = + tracker.snapshotSafeFrontiers(); + assertEquals(100L, before.get(7).getPhysicalTime()); + assertEquals(10L, before.get(7).getLocalSeq()); + + tracker.recordAppliedProgress(125L, 7, 20L); + final WriterSafeFrontierTracker.SafeFrontier after = tracker.snapshotSafeFrontiers().get(7); + assertEquals(130L, after.getPhysicalTime()); + assertEquals(20L, after.getLocalSeq()); + assertEquals(100L, before.get(7).getPhysicalTime()); + assertEquals(10L, before.get(7).getLocalSeq()); + } + @Test public void testPendingWriterSafeTimeBarrierPromotesWhenBarrierIsApplied() { final WriterSafeFrontierTracker tracker = new WriterSafeFrontierTracker(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java index dd5d2040009..fc657744e23 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 @@ -28,6 +28,7 @@ import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl; import org.apache.iotdb.consensus.iot.SubscriptionWalRetentionPolicy; +import org.apache.iotdb.consensus.iot.WriterSafeFrontierTracker.SafeFrontier; import org.apache.iotdb.consensus.iot.log.ConsensusReqReader; import org.apache.iotdb.consensus.iot.logdispatcher.IoTConsensusMemoryManager; import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueAdmission; @@ -361,6 +362,15 @@ public class ConsensusPrefetchingQueue { private final Map<Integer, NavigableMap<Long, PreparedEntry>> realtimeEntriesByWriter = new ConcurrentHashMap<>(); + // Ordered replay keeps one raw head per writer. Each iterator can pass other writers without + // converting or retaining their Tablets, so a long writer prefix cannot exhaust the memory + // needed to find an older record from another writer. + private final Map<Integer, WriterReplayState> orderedWriterReplay = new ConcurrentHashMap<>(); + private long orderedReplayStartSearchIndex = Long.MIN_VALUE; + private int orderedWriterRoundRobinStart; + private final AtomicLong orderedRequestBytes = new AtomicLong(); + private final Set<Long> reportedOrderedBrokenWalVersions = ConcurrentHashMap.newKeySet(); + /** * Local tail position used only when initialization starts without any persisted region progress. */ @@ -838,6 +848,7 @@ public class ConsensusPrefetchingQueue { } return hasImmediatePrefetchableWork() + || !orderedWriterReplay.isEmpty() || hasHistoricalWalLag() || !lingerBatch.isEmpty() || !inFlightEvents.isEmpty() @@ -1527,6 +1538,7 @@ public class ConsensusPrefetchingQueue { // ======================== Prefetch Round Drive ======================== private static final long WAL_GAP_RETRY_SLEEP_MS = 10L; + private static final long WRITER_BARRIER_RETRY_SLEEP_MS = 100L; private static final long MEMORY_RETRY_SLEEP_MS = 100L; private static final long WAL_GAP_WAIT_LOG_INTERVAL_MS = 5_000L; @@ -1636,6 +1648,16 @@ public class ConsensusPrefetchingQueue { final int maxTablets = config.getSubscriptionConsensusBatchMaxTabletCount(); final long maxBatchBytes = config.getSubscriptionConsensusBatchMaxSizeInBytes(); + if (!shouldUseActiveWriterBarriers() && !orderedWriterReplay.isEmpty()) { + closeOrderedWriterReplay(); + } + + if (shouldUseActiveWriterBarriers() + && (activeWriterNodeIds.size() > 1 || !orderedWriterReplay.isEmpty())) { + return driveOrderedWriterReplay( + observedSeekGeneration, maxWalEntries, maxTablets, maxBatchBytes, batchMaxDelayMs); + } + // Always consume already materialized entries before converting more WAL requests into // Tablets. Otherwise a full delivery batch can leave an unbounded hidden writer backlog. if (!drainBufferedRealtimeWriters( @@ -1647,6 +1669,7 @@ public class ConsensusPrefetchingQueue { || !realtimeEntriesByWriter.isEmpty()) { if (!realtimeEntriesByWriter.isEmpty()) { blockRealtimeAdmission(SubscriptionQueueRejectionReason.WRITER_BACKLOG); + return PrefetchRoundResult.rescheduleAfter(WRITER_BARRIER_RETRY_SLEEP_MS); } return computeIdleRoundResult(); } @@ -1765,6 +1788,348 @@ public class ConsensusPrefetchingQueue { } } + private PrefetchRoundResult driveOrderedWriterReplay( + final long expectedSeekGeneration, + final int maxWalEntries, + final int maxTablets, + final long maxBatchBytes, + final int batchMaxDelayMs) + throws IOException { + if (orderedReplayStartSearchIndex == Long.MIN_VALUE) { + orderedReplayStartSearchIndex = nextExpectedSearchIndex.get(); + if (!lingerBatch.isEmpty()) { + // A runtime transition can add another active writer while a single-writer batch lingers. + // Put the already reserved batch back into the merge before comparing the new writer. + bufferRealtimeEntry( + new PreparedEntry( + new ArrayList<>(lingerBatch.tablets), + lingerBatch.estimatedBytes, + lingerBatch.physicalTime, + lingerBatch.writerNodeId, + lingerBatch.lastLocalSeq, + lingerBatch.endSearchIndex)); + lingerBatch.reset(); + } + } + final Map<Integer, SafeFrontier> safeFrontiers = + serverImpl.getWriterSafeFrontierTracker().snapshotSafeFrontiers(); + activeWriterNodeIds.forEach(this::getOrCreateWriterReplay); + safeFrontiers.keySet().forEach(this::getOrCreateWriterReplay); + recoveryWriterProgressByWriter.keySet().forEach(id -> getOrCreateWriterReplay(id.getNodeId())); + realtimeEntriesByWriter.keySet().forEach(this::getOrCreateWriterReplay); + for (final IndexedConsensusRequest request : pendingEntries) { + if (hasComparableWriterProgress(request)) { + getOrCreateWriterReplay(request.getNodeId()); + } + } + + final int[] readBudget = {Math.max(1, maxWalEntries)}; + final int roundRobinStart = orderedWriterRoundRobinStart++; + while (prefetchingQueue.size() < MAX_PREFETCHING_QUEUE_SIZE) { + if (seekGeneration.get() != expectedSeekGeneration || closeRequested) { + return PrefetchRoundResult.rescheduleNow(); + } + final int writerCount = orderedWriterReplay.size(); + final List<Map.Entry<Integer, WriterReplayState>> writers = + new ArrayList<>(orderedWriterReplay.entrySet()); + Collections.rotate(writers, Math.floorMod(roundRobinStart, Math.max(1, writers.size()))); + for (final Map.Entry<Integer, WriterReplayState> writer : writers) { + if (!fillOrderedWriterHead(writer.getKey(), writer.getValue(), readBudget)) { + return flushOrderedBatchAndRetry(expectedSeekGeneration, MEMORY_RETRY_SLEEP_MS); + } + } + // A scan can discover a retired writer absent from the runtime set. Acquire its head before + // releasing any other writer, with the same shared per-round scan budget. + if (orderedWriterReplay.size() != writerCount && readBudget[0] > 0) { + continue; + } + + final PriorityQueue<WriterFrontier> frontiers = new PriorityQueue<>(); + for (final Map.Entry<Integer, WriterReplayState> writer : orderedWriterReplay.entrySet()) { + final int writerNodeId = writer.getKey(); + final WriterReplayState state = writer.getValue(); + final PreparedEntry buffered = peekRealtimeEntry(writerNodeId); + if (Objects.nonNull(buffered)) { + frontiers.add(WriterFrontier.forHead(writerNodeId, buffered)); + } else if (Objects.nonNull(state.head)) { + final IndexedConsensusRequest head = state.head; + frontiers.add( + new WriterFrontier( + writerNodeId, + new OrderingKey(head.getPhysicalTime(), writerNodeId, head.getProgressLocalSeq()), + false)); + } else if (isWriterRuntimeActive(writerNodeId) || !state.caughtUp) { + final SafeFrontier safe = safeFrontiers.get(writerNodeId); + final long safePt = + Objects.nonNull(safe) && state.consumedLocalSeq >= safe.getLocalSeq() + ? safe.getPhysicalTime() + : Long.MIN_VALUE; + frontiers.add(WriterFrontier.forBarrier(writerNodeId, safePt)); + } + } + final WriterFrontier next = frontiers.peek(); + if (Objects.isNull(next) || next.isBarrier) { + break; + } + final PreparedEntry buffered = peekRealtimeEntry(next.writerNodeId); + if (Objects.isNull(buffered) && shouldWaitForSubscriptionMemory()) { + blockRealtimeAdmission(memoryBlockReason); + return flushOrderedBatchAndRetry(expectedSeekGeneration, MEMORY_RETRY_SLEEP_MS); + } + final WriterReplayState state = orderedWriterReplay.get(next.writerNodeId); + final IndexedConsensusRequest request = state.head; + final PreparedEntry entry = Objects.nonNull(buffered) ? buffered : prepareEntry(request); + if (Objects.isNull(entry) && memoryBlockedEntryBytes > 0L) { + blockRealtimeAdmission(memoryBlockReason); + return flushOrderedBatchAndRetry(expectedSeekGeneration, MEMORY_RETRY_SLEEP_MS); + } + if (Objects.nonNull(entry)) { + if (!canAppendWriterEntry( + lingerBatch, + entry, + entry.estimatedBytes, + Integer.MAX_VALUE, + maxTablets, + maxBatchBytes) + && !flushBatch(lingerBatch, expectedSeekGeneration)) { + releaseTabletMemory(entry.estimatedBytes); + return PrefetchRoundResult.rescheduleNow(); + } + if (entry.tablets.isEmpty()) { + if (!commitEmptyWriterEntry(lingerBatch, entry, expectedSeekGeneration)) { + return PrefetchRoundResult.rescheduleNow(); + } + } else { + lingerBatch.append(entry, entry.estimatedBytes, true); + } + } + if (Objects.nonNull(buffered)) { + removeRealtimeEntry(buffered.writerNodeId, buffered.localSeq); + readBudget[0]--; + } else { + markMaterializedProgress(request); + state.consumedLocalSeq = Math.max(state.consumedLocalSeq, request.getProgressLocalSeq()); + if (state.fromPending) { + advanceLocalCursorIfPresent(request); + if (pendingEntries.remove(request)) { + pendingEntries.release(request); + } + markAcceptedFromPending(); + } else { + advanceWalReplayCursorIfPresent(request); + markAcceptedFromWal(); + } + releaseOrderedWriterHead(state); + } + unblockRealtimeAdmission(); + if (readBudget[0] <= 0) { + break; + } + } + + // Pending and WAL can carry the same record. Release obsolete realtime references even when + // this round used WAL, rather than letting their bounded queue fill with already read data. + for (final IndexedConsensusRequest request : pendingEntries) { + if (shouldSkipForRecoveryProgress(request) || shouldSkipForMaterializedProgress(request)) { + if (pendingEntries.remove(request)) { + pendingEntries.release(request); + } + } + } + if (!lingerBatch.isEmpty() + && System.currentTimeMillis() - lingerBatch.firstTabletTimeMs >= batchMaxDelayMs) { + if (!flushBatch(lingerBatch, expectedSeekGeneration)) { + return PrefetchRoundResult.rescheduleNow(); + } + } + maybeInjectWatermark(); + if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) { + return PrefetchRoundResult.dormant(); + } + if (readBudget[0] <= 0 + || (readBudget[0] < maxWalEntries + && orderedWriterReplay.values().stream() + .anyMatch(state -> Objects.isNull(state.head) && !state.caughtUp))) { + return PrefetchRoundResult.rescheduleNow(); + } + long delayMs = WRITER_BARRIER_RETRY_SLEEP_MS; + if (!lingerBatch.isEmpty()) { + delayMs = + Math.min( + delayMs, + Math.max( + 1L, + batchMaxDelayMs - (System.currentTimeMillis() - lingerBatch.firstTabletTimeMs))); + } + return PrefetchRoundResult.rescheduleAfter(delayMs); + } + + private PrefetchRoundResult flushOrderedBatchAndRetry( + final long expectedSeekGeneration, final long delayMs) { + if (!lingerBatch.isEmpty() && !flushBatch(lingerBatch, expectedSeekGeneration)) { + return PrefetchRoundResult.rescheduleNow(); + } + return PrefetchRoundResult.rescheduleAfter(delayMs); + } + + private WriterReplayState getOrCreateWriterReplay(final int writerNodeId) { + return orderedWriterReplay.computeIfAbsent( + writerNodeId, + ignored -> { + final WriterProgress recovered = + recoveryWriterProgressByWriter.get( + new WriterId(consensusGroupId.toString(), writerNodeId)); + final WriterProgress materialized = materializedProgressByWriter.get(writerNodeId); + return new WriterReplayState( + createSubscriptionWALIterator(Long.MIN_VALUE), + Math.max( + Objects.nonNull(recovered) ? recovered.getLocalSeq() : 0L, + Objects.nonNull(materialized) ? materialized.getLocalSeq() : 0L)); + }); + } + + private boolean fillOrderedWriterHead( + final int writerNodeId, final WriterReplayState state, final int[] readBudget) + throws IOException { + if (Objects.nonNull(state.head) || readBudget[0] <= 0) { + return true; + } + for (final IndexedConsensusRequest request : pendingEntries) { + if (shouldSkipForRecoveryProgress(request) + || shouldSkipForMaterializedProgress(request) + || (hasLocalSearchIndex(request) + && request.getSearchIndex() < orderedReplayStartSearchIndex)) { + if (pendingEntries.remove(request)) { + pendingEntries.release(request); + } + continue; + } + if (request.getNodeId() == writerNodeId + && !shouldSkipForRecoveryProgress(request) + && !shouldSkipForMaterializedProgress(request) + && (Objects.isNull(state.iterator) + || request.getProgressLocalSeq() == state.consumedLocalSeq + 1L)) { + state.head = request; + state.fromPending = true; + state.caughtUp = false; + readBudget[0]--; + return true; + } + } + if (Objects.isNull(state.iterator)) { + state.caughtUp = true; + return true; + } + state.iterator.skipReportedBrokenWalFiles(reportedOrderedBrokenWalVersions); + final int writerBudget = Math.max(1, readBudget[0] / Math.max(1, orderedWriterReplay.size())); + for (int scanned = 0; scanned < writerBudget && readBudget[0] > 0; scanned++) { + if (closeRequested || pendingSeekRequest != null) { + return true; + } + if (!state.iterator.hasNext()) { + state.iterator.refresh(); + if (!state.iterator.hasNext()) { + reportOrderedUnreadableWal(state.iterator); + state.caughtUp = + !state.iterator.hasReadError() && !state.iterator.isWaitingForReadableWal(); + return true; + } + } + reportOrderedUnreadableWal(state.iterator); + final IndexedConsensusRequest request = state.iterator.next(); + readBudget[0]--; + state.caughtUp = false; + if (hasComparableWriterProgress(request)) { + getOrCreateWriterReplay(request.getNodeId()); + } + if (request.getNodeId() != writerNodeId) { + continue; + } + if ((hasLocalSearchIndex(request) && request.getSearchIndex() < orderedReplayStartSearchIndex) + || shouldSkipForRecoveryProgress(request) + || shouldSkipForMaterializedProgress(request)) { + state.consumedLocalSeq = Math.max(state.consumedLocalSeq, request.getProgressLocalSeq()); + continue; + } + // WAL requests wrap ByteBuffers whose default getMemorySize() is zero. Build the buffer list + // before reserving the retained head so consensus accounting includes its actual payload. + request.buildSerializedRequests(); + if (!IoTConsensusMemoryManager.getInstance().reserve(request)) { + state.iterator.retry(request); + return false; + } + orderedRequestBytes.addAndGet(request.getRetainedMemorySize()); + state.head = request; + state.fromPending = false; + return true; + } + return true; + } + + private void reportOrderedUnreadableWal(final ProgressWALIterator iterator) { + int newFiles = 0; + for (final Map.Entry<Long, Long> skipped : + iterator.getSkippedBrokenWalEntryCounts().entrySet()) { + if (reportedOrderedBrokenWalVersions.add(skipped.getKey())) { + newFiles++; + unreadableWalEntryCount.addAndGet(skipped.getValue()); + } + } + if (newFiles > 0) { + final long totalSkippedFiles = unreadableWalFileCount.addAndGet(newFiles); + pendingWalReplayError.set( + String.format( + DataNodePipeMessages + .MESSAGE_SUBSCRIPTION_WAL_REPLAY_SKIPPED_ARG_UNREADABLE_RETAINED_WAL_FILE_S_IN_QUEUE_ARG_REGION_ARG_HISTORICAL_DATA_MAY_BE_LOST_DETAILS_RECENT_FILES_FULL_DETAILS_IN_SERVER_LOG_ARG_FEDEFF7E, + totalSkippedFiles, + getPrefetchingQueueId(), + consensusGroupId, + iterator.getSkippedBrokenWalFileDetails(0))); + } + } + + private void releaseOrderedWriterHead(final WriterReplayState state) { + if (Objects.nonNull(state.head) && !state.fromPending) { + orderedRequestBytes.addAndGet(-state.head.getRetainedMemorySize()); + IoTConsensusMemoryManager.getInstance().free(state.head); + } + state.head = null; + } + + private void closeOrderedWriterReplay() { + for (final WriterReplayState state : orderedWriterReplay.values()) { + releaseOrderedWriterHead(state); + if (Objects.nonNull(state.iterator)) { + try { + state.iterator.close(); + } catch (final IOException e) { + LOGGER.warn( + DataNodePipeMessages + .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_ERROR_CLOSING_SUBSCRIPTION_WAL_19711C01, + this, + e); + } + } + } + orderedWriterReplay.clear(); + reportedOrderedBrokenWalVersions.clear(); + orderedReplayStartSearchIndex = Long.MIN_VALUE; + orderedWriterRoundRobinStart = 0; + } + + private static final class WriterReplayState { + private final ProgressWALIterator iterator; + private long consumedLocalSeq; + private IndexedConsensusRequest head; + private boolean fromPending; + private boolean caughtUp; + + private WriterReplayState(final ProgressWALIterator iterator, final long consumedLocalSeq) { + this.iterator = iterator; + this.consumedLocalSeq = consumedLocalSeq; + } + } + private void logPeriodicStatsIfNecessary() { final long nowMs = System.currentTimeMillis(); if (nowMs - lastStatsLogTimeMs < PREFETCH_STATS_LOG_INTERVAL_MS) { @@ -1825,6 +2190,9 @@ public class ConsensusPrefetchingQueue { return PrefetchRoundResult.rescheduleNow(); } long delayMs = Long.MAX_VALUE; + if (!realtimeEntriesByWriter.isEmpty()) { + delayMs = Math.min(delayMs, WRITER_BARRIER_RETRY_SLEEP_MS); + } if (hasHistoricalWalLag()) { delayMs = Math.min(delayMs, WAL_GAP_RETRY_SLEEP_MS); } @@ -1869,9 +2237,10 @@ public class ConsensusPrefetchingQueue { } private boolean hasImmediatePrefetchableWork() { - return !pendingEntries.isEmpty() - || !realtimeEntriesByWriter.isEmpty() - || hasReadableWalEntries(); + if (!orderedWriterReplay.isEmpty()) { + return false; + } + return !pendingEntries.isEmpty() || hasReadableWalEntries(); } private boolean hasHistoricalWalLag() { @@ -2370,6 +2739,7 @@ public class ConsensusPrefetchingQueue { } private void closeSubscriptionWALIterator() { + closeOrderedWriterReplay(); if (Objects.isNull(subscriptionWALIterator)) { return; } @@ -4771,7 +5141,7 @@ public class ConsensusPrefetchingQueue { } public long getRetainedRequestBytes() { - return pendingEntries.getRetainedRequestBytes(); + return pendingEntries.getRetainedRequestBytes() + orderedRequestBytes.get(); } public long getSubscriptionMemoryLimitInBytes() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java index 3636829274f..ed859e6d3dc 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java @@ -45,9 +45,12 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; import java.util.List; +import java.util.Map; import java.util.NoSuchElementException; import java.util.Objects; import java.util.Set; @@ -86,6 +89,7 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus private boolean currentReaderUsesLiveSnapshot = false; private int consumedEntryCountInCurrentFile = 0; private final Set<Long> skippedBrokenWalVersionIds = new HashSet<>(); + private final Map<Long, Long> skippedBrokenWalEntryCounts = new HashMap<>(); private final List<String> skippedBrokenWalFileDetails = new ArrayList<>(); private long skippedBrokenWalEntryCount; private boolean retryCurrentReader; @@ -393,6 +397,32 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus return skippedBrokenWalEntryCount; } + Map<Long, Long> getSkippedBrokenWalEntryCounts() { + return Collections.unmodifiableMap(skippedBrokenWalEntryCounts); + } + + /** Shares confirmed corrupt-file decisions between independent readers of the same WAL. */ + void skipReportedBrokenWalFiles(final Set<Long> versionIds) throws IOException { + final boolean skipsCurrentFile = + versionIds.contains(currentReaderVersionId) + || (currentFileIndex >= 0 + && currentFileIndex < walFileVersionIds.length + && versionIds.contains(walFileVersionIds[currentFileIndex])); + if (skipsCurrentFile) { + closeCurrentReader(); + nextReady = null; + pendingRequests.clear(); + pendingSearchIndex = Long.MIN_VALUE; + pendingLocalSeq = Long.MIN_VALUE; + } + if (skippedBrokenWalVersionIds.addAll(versionIds)) { + refresh(); + } + if (skipsCurrentFile) { + resetCurrentFileTracking(); + } + } + boolean isWaitingForReadableWal() { return retryCurrentReader; } @@ -703,6 +733,7 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus final int entryCount = metadata == null ? -1 : metadata.getBuffersSize().size(); final long skippedEntries = entryCount < 0 ? -1L : Math.max(0L, (long) entryCount - firstSkippedEntryOffset); + skippedBrokenWalEntryCounts.put(versionId, Math.max(0L, skippedEntries)); if (skippedEntries > 0L) { skippedBrokenWalEntryCount += skippedEntries; } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDataNodeMemoryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDataNodeMemoryTest.java index 16f675a4da2..d6523250872 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDataNodeMemoryTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDataNodeMemoryTest.java @@ -510,7 +510,7 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest { } @Test - public void testRealtimeBacklogIsDrainedBeforeMorePendingEntriesAreConverted() throws Exception { + public void testWriterBarrierDoesNotMaterializeBlockedPendingEntries() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); final int originalBatchMaxDelay = CommonDescriptor.getInstance().getConfig().getSubscriptionConsensusBatchMaxDelayInMs(); @@ -538,21 +538,19 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest { assertTrue(pendingEntries(queue).offer(createRequest(1L))); queue.drivePrefetchOnce(); - assertEquals(1, conversionCount.get()); - assertEquals(2L, queue.getCurrentReadSearchIndex()); - assertEquals("1", queue.coreReportMessage().get("bufferedRealtimeEntryCount")); + assertEquals(0, conversionCount.get()); + assertEquals(1L, queue.getCurrentReadSearchIndex()); + assertEquals("0", queue.coreReportMessage().get("bufferedRealtimeEntryCount")); assertEquals(0, queue.getPrefetchedEventCount()); assertTrue(pendingEntries(queue).offer(createRequest(2L))); queue.drivePrefetchOnce(); - assertEquals(1, conversionCount.get()); - assertEquals(2L, queue.getCurrentReadSearchIndex()); - assertEquals("1", queue.coreReportMessage().get("bufferedRealtimeEntryCount")); - assertEquals("1", queue.coreReportMessage().get("pendingEntriesSize")); - assertEquals("true", queue.coreReportMessage().get("realtimeAdmissionBlocked")); - assertFalse(pendingEntries(queue).offer(createRequest(3L))); - assertEquals("SUBSCRIPTION_WRITER_BACKLOG", queue.getLastAdmissionRejectionCode()); + assertEquals(0, conversionCount.get()); + assertEquals(1L, queue.getCurrentReadSearchIndex()); + assertEquals("0", queue.coreReportMessage().get("bufferedRealtimeEntryCount")); + assertEquals("2", queue.coreReportMessage().get("pendingEntriesSize")); + assertEquals(0L, queue.getRetainedTabletBytes()); assertEquals(0L, queue.getSubscriptionMemoryRejectionCount()); queue.setActiveWriterNodeIds(Collections.singleton(7)); 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 54fcb704f00..dd75483599d 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 @@ -57,6 +57,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.Iterator; import java.util.List; +import java.util.Set; import java.util.concurrent.TimeUnit; import static org.junit.Assert.assertEquals; @@ -91,11 +92,26 @@ public class ConsensusPrefetchingQueueSeekTest { Arrays.asList(createRequest(3L, 3L, 7), createRequest(4L, 4L, 7)), 5L, 2L); } + @Test(timeout = 10_000L) + public void testMultiWriterSeekReplaysAfterResettingOrderedReaders() throws Exception { + assertBeginningReplay( + Arrays.asList(createRequest(1L, 1L, 7), createRequest(-1L, 10L, 8)), 2L, 0L, true); + } + private void assertBeginningReplay( final List<IndexedConsensusRequest> requests, final long expectedNextIndex, final long expectedSkippedEntries) throws Exception { + assertBeginningReplay(requests, expectedNextIndex, expectedSkippedEntries, false); + } + + private void assertBeginningReplay( + final List<IndexedConsensusRequest> requests, + final long expectedNextIndex, + final long expectedSkippedEntries, + final boolean multiWriter) + throws Exception { final CommonConfig config = CommonDescriptor.getInstance().getConfig(); final int originalBatchMaxDelay = config.getSubscriptionConsensusBatchMaxDelayInMs(); final int originalPrefetchThreads = @@ -124,7 +140,14 @@ public class ConsensusPrefetchingQueueSeekTest { when(walNode.getLogDirectory()).thenReturn(systemDir); final IoTConsensusServerImpl server = mock(IoTConsensusServerImpl.class); when(server.getConsensusReqReader()).thenReturn(walNode); - when(server.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + final WriterSafeFrontierTracker tracker = new WriterSafeFrontierTracker(); + for (final IndexedConsensusRequest request : requests) { + tracker.recordAppliedProgress( + request.getPhysicalTime(), request.getNodeId(), request.getProgressLocalSeq()); + tracker.observePendingWriterSafeTimeBarrier( + 3000L, request.getNodeId(), request.getProgressLocalSeq()); + } + when(server.getWriterSafeFrontierTracker()).thenReturn(tracker); final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); when(converter.getDatabaseName()).thenReturn("db"); when(converter.isTableModel()).thenReturn(true); @@ -138,7 +161,9 @@ public class ConsensusPrefetchingQueueSeekTest { new ConsensusPrefetchingQueue( "consumerGroup", "topic", - TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + multiWriter + ? TopicConstant.ORDER_MODE_MULTI_WRITER_VALUE + : TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, new DataRegionId(1), server, new SubscriptionWalRetentionPolicy( @@ -170,36 +195,41 @@ public class ConsensusPrefetchingQueueSeekTest { } }; queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); + if (multiWriter) { + queue.setActiveWriterNodeIds(Set.of(7, 8)); + } - queue.seekToBeginning(); + for (int attempt = 0; attempt < (multiWriter ? 2 : 1); attempt++) { + queue.seekToBeginning(); - final List<Long> actualTimestamps = new ArrayList<>(); - final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5L); - while (actualTimestamps.size() < requests.size() && System.nanoTime() < deadline) { - final SubscriptionEvent event = queue.poll("consumer"); - if (event == null) { - TimeUnit.MILLISECONDS.sleep(10L); - continue; + final List<Long> actualTimestamps = new ArrayList<>(); + final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5L); + while (actualTimestamps.size() < requests.size() && System.nanoTime() < deadline) { + final SubscriptionEvent event = queue.poll("consumer"); + if (event == null) { + TimeUnit.MILLISECONDS.sleep(10L); + continue; + } + assertEquals( + SubscriptionPollResponseType.TABLETS.getType(), + event.getCurrentResponse().getResponseType()); + for (final Tablet tablet : + ((TabletsPayload) event.getCurrentResponse().getPayload()).getTablets()) { + assertEquals(1, tablet.getRowSize()); + actualTimestamps.add(tablet.getTimestamps()[0]); + } } - assertEquals( - SubscriptionPollResponseType.TABLETS.getType(), - event.getCurrentResponse().getResponseType()); - for (final Tablet tablet : - ((TabletsPayload) event.getCurrentResponse().getPayload()).getTablets()) { - assertEquals(1, tablet.getRowSize()); - actualTimestamps.add(tablet.getTimestamps()[0]); + final List<Long> expectedTimestamps = new ArrayList<>(); + for (final IndexedConsensusRequest request : requests) { + expectedTimestamps.add(((InsertRowNode) request.getRequests().get(0)).getTime()); } + Collections.sort(expectedTimestamps); + Collections.sort(actualTimestamps); + assertEquals(expectedTimestamps, actualTimestamps); + assertEquals(expectedNextIndex, queue.getCurrentReadSearchIndex()); + assertEquals(expectedSkippedEntries, queue.getWalGapSkippedEntries()); + verify(converter, times(requests.size() * (attempt + 1))).convert(any()); } - final List<Long> expectedTimestamps = new ArrayList<>(); - for (final IndexedConsensusRequest request : requests) { - expectedTimestamps.add(((InsertRowNode) request.getRequests().get(0)).getTime()); - } - Collections.sort(expectedTimestamps); - Collections.sort(actualTimestamps); - assertEquals(expectedTimestamps, actualTimestamps); - assertEquals(expectedNextIndex, queue.getCurrentReadSearchIndex()); - assertEquals(expectedSkippedEntries, queue.getWalGapSkippedEntries()); - verify(converter, times(requests.size())).convert(any()); } finally { if (queue != null) { queue.close(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWriterOrderingTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWriterOrderingTest.java new file mode 100644 index 00000000000..8cca9e99925 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWriterOrderingTest.java @@ -0,0 +1,484 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.subscription.broker.consensus; + +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.consensus.DataRegionId; +import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; +import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl; +import org.apache.iotdb.consensus.iot.SubscriptionWalRetentionPolicy; +import org.apache.iotdb.consensus.iot.WriterSafeFrontierTracker; +import org.apache.iotdb.consensus.iot.log.ConsensusReqReader; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; +import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFileVersion; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALMetaData; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALWriter; +import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileStatus; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils; +import org.apache.iotdb.db.subscription.event.SubscriptionEvent; +import org.apache.iotdb.db.subscription.resource.SubscriptionMemoryManager; +import org.apache.iotdb.rpc.subscription.config.TopicConstant; +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.tsfile.enums.TSDataType; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.io.File; +import java.lang.reflect.Field; +import java.nio.ByteBuffer; +import java.nio.file.Files; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; + +public class ConsensusPrefetchingQueueWriterOrderingTest { + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private final CommonConfig config = CommonDescriptor.getInstance().getConfig(); + private String originalSystemDir; + private int originalWalEntries; + private int originalTablets; + private int originalDelay; + private final WriterSafeFrontierTracker tracker = new WriterSafeFrontierTracker(); + private final AtomicInteger conversions = new AtomicInteger(); + private final AtomicInteger walReads = new AtomicInteger(); + private ConsensusPrefetchingQueue queue; + + @Before + public void setUp() throws Exception { + originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + originalWalEntries = config.getSubscriptionConsensusBatchMaxWalEntries(); + originalTablets = config.getSubscriptionConsensusBatchMaxTabletCount(); + originalDelay = config.getSubscriptionConsensusBatchMaxDelayInMs(); + config.setSubscriptionConsensusBatchMaxWalEntries(1024); + config.setSubscriptionConsensusBatchMaxTabletCount(512); + config.setSubscriptionConsensusBatchMaxDelayInMs(0); + IoTDBDescriptor.getInstance() + .getConfig() + .setSystemDir(temporaryFolder.newFolder("system").getAbsolutePath()); + } + + @After + public void tearDown() { + if (queue != null) { + queue.close(); + assertEquals(0L, queue.getRetainedRequestBytes()); + assertEquals(0L, queue.getRetainedTabletBytes()); + } + config.setSubscriptionConsensusBatchMaxWalEntries(originalWalEntries); + config.setSubscriptionConsensusBatchMaxTabletCount(originalTablets); + config.setSubscriptionConsensusBatchMaxDelayInMs(originalDelay); + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + + @Test + public void testBlockedWriterBarrierRetriesLaterAndResumesAfterIdleBarrier() throws Exception { + newQueue(mock(ConsensusReqReader.class), 1024 * 1024); + tracker.recordAppliedProgress(2000, 7, 1); + tracker.observePendingWriterSafeTimeBarrier(1000, 8, 0); + assertTrue(pending().offer(request(1, 7, 2000))); + for (int round = 0; round < 100; round++) { + final PrefetchRoundResult result = queue.drivePrefetchOnce(); + assertEquals(PrefetchRoundResult.Type.RESCHEDULE_LATER, result.getType()); + assertTrue(result.getDelayMs() > 0); + } + assertEquals(0, conversions.get()); + assertEquals(0L, queue.getRetainedTabletBytes()); + tracker.observePendingWriterSafeTimeBarrier(3000, 8, 0); + queue.drivePrefetchOnce(); + assertEvent(7, 2000); + } + + @Test + public void testAppliedBarrierCannotHideAnUnreadWriter() throws Exception { + newQueue(mock(ConsensusReqReader.class), 1024 * 1024); + installSafeFrontiers(1); + assertTrue(pending().offer(request(1, 7, 2000))); + assertEquals(PrefetchRoundResult.Type.RESCHEDULE_LATER, queue.drivePrefetchOnce().getType()); + assertEquals(0, conversions.get()); + assertNull(queue.poll("consumer")); + assertTrue(pending().offer(request(-1, 8, 1000))); + queue.drivePrefetchOnce(); + assertEvent(8, 1000); + queue.drivePrefetchOnce(); + assertEvent(7, 2000); + } + + @Test + public void testDefaultPendingBatchBoundaryKeepsWriterOrdering() throws Exception { + config.setSubscriptionConsensusBatchMaxDelayInMs(50); + newQueue(mock(ConsensusReqReader.class), 16 * 1024 * 1024); + installSafeFrontiers(1024); + for (int index = 1; index <= 1024; index++) { + assertTrue(pending().offer(request(index, 7, 2000))); + } + assertTrue(pending().offer(request(-1, 8, 1000))); + queue.drivePrefetchOnce(); + assertEvent(8, 1000); + consumeRemainingWriterSeven(1024); + } + + @Test + public void testPendingOrderingFitsOneTabletMemoryBudget() throws Exception { + newQueue(mock(ConsensusReqReader.class), oneTabletBudget()); + installSafeFrontiers(1024); + for (int index = 1; index <= 1024; index++) { + assertTrue(pending().offer(request(index, 7, 2000))); + } + assertTrue(pending().offer(request(-1, 8, 1000))); + queue.drivePrefetchOnce(); + assertEvent(8, 1000); + consumeRemainingWriterSeven(1024); + } + + @Test + public void testWalOrderingReadsOlderWriterBeyondRoundBudgetWithoutTabletBacklog() + throws Exception { + final File directory = writeWal(1024); + final WALNode walNode = walNode(directory, 1024); + config.setSubscriptionConsensusBatchMaxWalEntries(32); + newQueue(walNode, oneTabletBudget()); + installSafeFrontiers(1024); + queue.drivePrefetchOnce(); + assertEquals(0, conversions.get()); + assertEquals(0L, queue.getRetainedTabletBytes()); + SubscriptionEvent first = null; + for (int round = 0; round < 500 && first == null; round++) { + final int readsBefore = walReads.get(); + queue.drivePrefetchOnce(); + assertTrue(walReads.get() - readsBefore <= 32); + first = queue.poll("consumer"); + } + assertNotNull(first); + assertEquals(8, first.getCommitContext().getWriterId().getNodeId()); + assertEquals(1000L, first.getCommitContext().getWriterProgress().getPhysicalTime()); + assertTrue(queue.ack("consumer", first.getCommitContext())); + consumeRemainingWriterSeven(1024); + } + + @Test + public void testDeactivationReleasesRawWriterHeads() throws Exception { + newQueue(walNode(writeWal(1), 1), oneTabletBudget()); + tracker.recordAppliedProgress(2000, 7, 1); + tracker.observePendingWriterSafeTimeBarrier(500, 8, 0); + queue.drivePrefetchOnce(); + assertTrue(queue.getRetainedRequestBytes() > 0); + assertEquals(1, conversions.get()); + queue.setActive(false); + assertEquals(0L, queue.getRetainedRequestBytes()); + assertEquals(0L, queue.getRetainedTabletBytes()); + } + + @Test + public void testAppliedWriterWaitsForWalVisibilityAndResumesOnRefresh() throws Exception { + final File directory = temporaryFolder.newFolder("delayed-wal"); + final WALNode walNode = walNode(directory, 1); + newQueue(walNode, oneTabletBudget()); + installSafeFrontiers(1); + assertTrue(pending().offer(request(1, 7, 2000))); + for (int round = 0; round < 10; round++) { + assertEquals(PrefetchRoundResult.Type.RESCHEDULE_LATER, queue.drivePrefetchOnce().getType()); + } + assertEquals(0, conversions.get()); + assertEquals(0L, queue.getRetainedTabletBytes()); + final File data = + new File(directory, WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)); + try (WALWriter writer = new WALWriter(data, WALFileVersion.V3)) { + writeEntry(writer, 1, 7, 2000, 1); + writeEntry(writer, -1, 8, 1000, 1); + } + try (WALWriter ignored = + new WALWriter( + new File( + directory, WALFileUtils.getLogFileName(1, 1, WALFileStatus.CONTAINS_SEARCH_INDEX)), + WALFileVersion.V3)) { + // The independent readers must refresh after previously exhausting an empty directory. + } + queue.drivePrefetchOnce(); + assertEvent(8, 1000); + queue.drivePrefetchOnce(); + assertEvent(7, 2000); + assertNull(queue.poll("consumer")); + } + + @Test + public void testWalAndPendingCopiesAreMaterializedOnce() throws Exception { + newQueue(walNode(writeWal(2), 2), 1024 * 1024); + installSafeFrontiers(2); + assertTrue(pending().offer(request(1, 7, 2000))); + assertTrue(pending().offer(request(2, 7, 2000))); + assertTrue(pending().offer(request(-1, 8, 1000))); + queue.drivePrefetchOnce(); + assertEvent(8, 1000); + consumeRemainingWriterSeven(2); + for (int round = 0; round < 10; round++) { + queue.drivePrefetchOnce(); + } + assertEquals(3, conversions.get()); + assertEquals(0, queue.getPendingEventCount()); + assertEquals(0, queue.getPrefetchedEventCount()); + } + + @Test + public void testOrderedReadersReportBrokenWalFileOnlyOnce() throws Exception { + final File directory = temporaryFolder.newFolder("broken-wal"); + final File broken = + new File(directory, WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)); + Files.write(broken.toPath(), new byte[128]); + try (WALWriter writer = + new WALWriter( + new File( + directory, WALFileUtils.getLogFileName(1, 1, WALFileStatus.CONTAINS_SEARCH_INDEX)), + WALFileVersion.V3)) { + writeEntry(writer, 2, 7, 2000, 2); + writeEntry(writer, -1, 8, 1000, 1); + } + try (WALWriter ignored = + new WALWriter( + new File( + directory, WALFileUtils.getLogFileName(2, 2, WALFileStatus.CONTAINS_SEARCH_INDEX)), + WALFileVersion.V3)) { + // Keep both the unreadable prefix and the intact data file historical. + } + final WALNode walNode = walNode(directory, 2); + when(walNode.getCurrentWALFileVersion()).thenReturn(2L); + newQueue(walNode, 1024 * 1024); + installSafeFrontiers(2); + // Retention initialization may recover an unreadable footer. Corrupt the sealed prefix after + // initialization to exercise the subscription readers' error reporting, rather than recovery. + Files.write(broken.toPath(), new byte[128]); + queue.drivePrefetchOnce(); + final SubscriptionEvent error = queue.poll("consumer"); + assertNotNull(error); + assertEquals( + SubscriptionPollResponseType.ERROR.getType(), error.getCurrentResponse().getResponseType()); + assertEquals(1L, queue.getUnreadableWalFileCount()); + queue.drivePrefetchOnce(); + assertEvent(8, 1000); + queue.drivePrefetchOnce(); + assertEvent(7, 2000); + queue.drivePrefetchOnce(); + assertEquals(1L, queue.getUnreadableWalFileCount()); + assertEquals(1L, queue.getWalGapSkippedEntries()); + assertNull(queue.poll("consumer")); + } + + private void consumeRemainingWriterSeven(final int expectedCount) throws Exception { + int tablets = 0; + long lastSeq = 0; + for (int round = 0; round < expectedCount * 4 + 100 && tablets < expectedCount; round++) { + final int readsBefore = walReads.get(); + queue.drivePrefetchOnce(); + assertTrue( + walReads.get() - readsBefore <= config.getSubscriptionConsensusBatchMaxWalEntries()); + SubscriptionEvent event; + while ((event = queue.poll("consumer")) != null) { + assertEquals(7, event.getCommitContext().getWriterId().getNodeId()); + assertEquals(2000L, event.getCommitContext().getWriterProgress().getPhysicalTime()); + assertTrue(event.getCommitContext().getWriterProgress().getLocalSeq() > lastSeq); + lastSeq = event.getCommitContext().getWriterProgress().getLocalSeq(); + tablets += ((TabletsPayload) event.getCurrentResponse().getPayload()).getTablets().size(); + assertTrue(queue.ack("consumer", event.getCommitContext())); + } + assertTrue(queue.getRetainedTabletBytes() <= queue.getSubscriptionMemoryLimitInBytes()); + if (tablets < expectedCount && config.getSubscriptionConsensusBatchMaxDelayInMs() > 0) { + Thread.sleep(60); + } + } + assertEquals(expectedCount, tablets); + assertEquals(expectedCount, lastSeq); + assertEquals(0L, queue.getRetainedTabletBytes()); + } + + private void assertEvent(final int writerNodeId, final long physicalTime) { + final SubscriptionEvent event = queue.poll("consumer"); + assertNotNull(queue.coreReportMessage().toString(), event); + assertEquals(writerNodeId, event.getCommitContext().getWriterId().getNodeId()); + assertEquals(physicalTime, event.getCommitContext().getWriterProgress().getPhysicalTime()); + assertTrue(queue.ack("consumer", event.getCommitContext())); + } + + private void installSafeFrontiers(final long writerSevenSeq) { + tracker.recordAppliedProgress(2000, 7, writerSevenSeq); + tracker.recordAppliedProgress(1000, 8, 1); + tracker.observePendingWriterSafeTimeBarrier(3000, 7, writerSevenSeq); + tracker.observePendingWriterSafeTimeBarrier(3000, 8, 1); + } + + private void newQueue(final ConsensusReqReader reader, final long memoryBytes) { + final IoTConsensusServerImpl server = mock(IoTConsensusServerImpl.class); + when(server.getConsensusReqReader()).thenReturn(reader); + when(server.getWriterSafeFrontierTracker()).thenReturn(tracker); + final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); + when(converter.getDatabaseName()).thenReturn("db"); + when(converter.convert(any())) + .thenAnswer( + ignored -> { + conversions.incrementAndGet(); + return Collections.singletonList(tablet()); + }); + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + "topic", + TopicConstant.ORDER_MODE_MULTI_WRITER_VALUE, + new DataRegionId(1), + server, + new SubscriptionWalRetentionPolicy( + "topic", + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + converter, + new ConsensusSubscriptionCommitManager( + (group, topic, id) -> + ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.absent()), + new RegionProgress(Collections.emptyMap()), + 1, + 1, + true) { + @Override + protected ProgressWALIterator createSubscriptionWALIterator(final long startSearchIndex) { + final ProgressWALIterator iterator = + super.createSubscriptionWALIterator(startSearchIndex); + if (iterator == null) { + return null; + } + final ProgressWALIterator tracked = spy(iterator); + doAnswer( + invocation -> { + walReads.incrementAndGet(); + return invocation.callRealMethod(); + }) + .when(tracked) + .next(); + return tracked; + } + }; + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(memoryBytes)); + queue.setActiveWriterNodeIds(Set.of(7, 8)); + assertNull(queue.poll("consumer")); + } + + private static long oneTabletBudget() { + final long bytes = PipeMemoryWeightUtil.calculateTabletSizeInBytes(tablet()); + return bytes + bytes / 2; + } + + private static Tablet tablet() { + final Tablet tablet = + new Tablet("root.test.device", List.of(new MeasurementSchema("s", TSDataType.INT64)), 1); + tablet.addTimestamp(0, 1); + tablet.addValue("s", 0, 1L); + tablet.setRowSize(1); + return tablet; + } + + private static IndexedConsensusRequest request( + final long searchIndex, final int writerNodeId, final long physicalTime) { + return new IndexedConsensusRequest( + searchIndex, + writerNodeId == 8 ? 1 : searchIndex, + Collections.singletonList(StatementTestUtils.genInsertRowNode(1))) + .setNodeId(writerNodeId) + .setPhysicalTime(physicalTime); + } + + @SuppressWarnings("unchecked") + private BlockingQueue<IndexedConsensusRequest> pending() throws Exception { + final Field field = ConsensusPrefetchingQueue.class.getDeclaredField("pendingEntries"); + field.setAccessible(true); + return (BlockingQueue<IndexedConsensusRequest>) field.get(queue); + } + + private static WALNode walNode(final File directory, final long currentIndex) { + final WALNode walNode = mock(WALNode.class); + when(walNode.getLogDirectory()).thenReturn(directory); + when(walNode.getCurrentSearchIndex()).thenReturn(currentIndex); + when(walNode.getCurrentWALFileVersion()).thenReturn(1L); + when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(new WALMetaData()); + return walNode; + } + + private File writeWal(final int localCount) throws Exception { + final File directory = temporaryFolder.newFolder("wal"); + final File data = + new File(directory, WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)); + try (WALWriter writer = new WALWriter(data, WALFileVersion.V3)) { + for (int index = 1; index <= localCount; index++) { + writeEntry(writer, index, 7, 2000, index); + } + writeEntry(writer, -1, 8, 1000, 1); + } + try (WALWriter ignored = + new WALWriter( + new File( + directory, + WALFileUtils.getLogFileName(1, localCount, WALFileStatus.CONTAINS_SEARCH_INDEX)), + WALFileVersion.V3)) { + // Seal the data file while keeping the live successor empty. + } + return directory; + } + + private static void writeEntry( + final WALWriter writer, + final long searchIndex, + final int writerNodeId, + final long physicalTime, + final long localSeq) + throws Exception { + final RelationalInsertRowNode node = StatementTestUtils.genInsertRowNode(1); + node.setSearchIndex(searchIndex); + node.setLastFragment(true); + final WALInfoEntry entry = new WALInfoEntry(1, node); + final ByteBuffer buffer = ByteBuffer.allocate(entry.serializedSize()); + entry.serialize(new WALByteBufferForTest(buffer)); + final WALMetaData metadata = new WALMetaData(); + metadata.add(entry.serializedSize(), searchIndex, 1, physicalTime, writerNodeId, localSeq); + writer.write(buffer, metadata); + } +}
