This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 4c34a253927 [Subscription] Preserve writer order and back off on
blocked safe frontiers (#18844)
4c34a253927 is described below
commit 4c34a253927be78118d19b7b31f33b2f832d5453
Author: Caideyipi <[email protected]>
AuthorDate: Mon Oct 19 15:36:49 2026 +0800
[Subscription] Preserve writer order and back off on blocked safe frontiers
(#18844)
* [Subscription] Preserve writer order and back off on blocked safe
frontiers
* Clarify ordered WAL scan budget accounting
---
.../consensus/iot/WriterSafeFrontierTracker.java | 33 ++
.../iot/WriterSafeFrontierTrackerTest.java | 20 +
.../consensus/ConsensusPrefetchingQueue.java | 407 +++++++++++++++-
.../broker/consensus/ProgressWALIterator.java | 31 ++
...onsensusPrefetchingQueueDataNodeMemoryTest.java | 21 +-
.../ConsensusPrefetchingQueueDeliveryTest.java | 42 +-
.../ConsensusPrefetchingQueueSeekTest.java | 84 ++--
...onsensusPrefetchingQueueWriterOrderingTest.java | 521 +++++++++++++++++++++
8 files changed, 1118 insertions(+), 41 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 ae71c4fea2a..5d0443c6996 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;
@@ -391,6 +392,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.
*/
@@ -945,6 +955,7 @@ public class ConsensusPrefetchingQueue {
}
return hasImmediatePrefetchableWork()
+ || !orderedWriterReplay.isEmpty()
|| hasHistoricalWalLag()
|| !lingerBatch.isEmpty()
|| !inFlightEvents.isEmpty()
@@ -1662,6 +1673,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;
// Yield long WAL scans to other regions without closing readers or
splitting writer requests.
private static final long WAL_SCAN_TIME_SLICE_NANOS =
TimeUnit.MILLISECONDS.toNanos(50L);
@@ -1778,6 +1790,21 @@ 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,
+ walScanBudget);
+ }
+
// 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(
@@ -1920,6 +1947,374 @@ public class ConsensusPrefetchingQueue {
}
}
+ private PrefetchRoundResult driveOrderedWriterReplay(
+ final long expectedSeekGeneration,
+ final int maxWalEntries,
+ final int maxTablets,
+ final long maxBatchBytes,
+ final int batchMaxDelayMs,
+ final ProgressWALIterator.ScanBudget walScanBudget)
+ 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.
+ // Preserve each entry's ordering key, WAL position, and projection
when returning the
+ // reserved Tablets to the merge, so the resulting delivery remains
replayable.
+ for (final WriterBufferedEntry entry : lingerBatch.entries) {
+ bufferRealtimeEntry(
+ new PreparedEntry(
+ entry.getTablets(),
+ entry.getEstimatedBytes(),
+ entry.getPhysicalTime(),
+ entry.getWriterNodeId(),
+ entry.getLocalSeq(),
+ entry.getSearchIndex(),
+ entry.getReplayConverter()));
+ }
+ 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());
+ }
+ }
+
+ // Bound entries processed across all writer readers in this round,
including skipped entries.
+ final int[] remainingEntryBudget = {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(), remainingEntryBudget,
walScanBudget)) {
+ return flushOrderedBatchAndRetry(expectedSeekGeneration,
MEMORY_RETRY_SLEEP_MS);
+ }
+ if (walScanBudget.hasYielded()) {
+ break;
+ }
+ }
+ // 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 && remainingEntryBudget[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);
+ remainingEntryBudget[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 (remainingEntryBudget[0] <= 0) {
+ break;
+ }
+ if (walScanBudget.hasYielded()) {
+ 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 (walScanBudget.hasYielded()
+ || remainingEntryBudget[0] <= 0
+ || (remainingEntryBudget[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[] remainingEntryBudget,
+ final ProgressWALIterator.ScanBudget walScanBudget)
+ throws IOException {
+ if (Objects.nonNull(state.head) || remainingEntryBudget[0] <= 0 ||
walScanBudget.hasYielded()) {
+ 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;
+ remainingEntryBudget[0]--;
+ return true;
+ }
+ }
+ if (Objects.isNull(state.iterator)) {
+ state.caughtUp = true;
+ return true;
+ }
+
state.iterator.skipReportedBrokenWalFiles(reportedOrderedBrokenWalVersions);
+ final int writerEntryBudget =
+ Math.max(1, remainingEntryBudget[0] / Math.max(1,
orderedWriterReplay.size()));
+ for (int scanned = 0; scanned < writerEntryBudget &&
remainingEntryBudget[0] > 0; scanned++) {
+ if (closeRequested || pendingSeekRequest != null) {
+ return true;
+ }
+ if (!state.iterator.hasNext(walScanBudget)) {
+ if (walScanBudget.hasYielded()) {
+ return true;
+ }
+ state.iterator.refresh();
+ if (!state.iterator.hasNext(walScanBudget)) {
+ if (walScanBudget.hasYielded()) {
+ return true;
+ }
+ reportOrderedUnreadableWal(state.iterator);
+ state.caughtUp =
+ !state.iterator.hasReadError() &&
!state.iterator.isWaitingForReadableWal();
+ return true;
+ }
+ }
+ reportOrderedUnreadableWal(state.iterator);
+ final IndexedConsensusRequest request = state.iterator.next();
+ // Each reader decodes the shared WAL. Charge skipped and duplicate
entries too so a long
+ // prefix cannot monopolize this round's prefetch work.
+ remainingEntryBudget[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) {
@@ -2022,6 +2417,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);
}
@@ -2066,6 +2464,9 @@ public class ConsensusPrefetchingQueue {
}
private boolean hasImmediatePrefetchableWork() {
+ if (!orderedWriterReplay.isEmpty()) {
+ return false;
+ }
return hasImmediatePrefetchableWork(newWalScanBudget());
}
@@ -2630,6 +3031,7 @@ public class ConsensusPrefetchingQueue {
}
private void closeSubscriptionWALIterator() {
+ closeOrderedWriterReplay();
if (Objects.isNull(subscriptionWALIterator)) {
return;
}
@@ -5265,7 +5667,7 @@ public class ConsensusPrefetchingQueue {
}
public long getRetainedRequestBytes() {
- return pendingEntries.getRetainedRequestBytes();
+ return pendingEntries.getRetainedRequestBytes() +
orderedRequestBytes.get();
}
public long getSubscriptionMemoryLimitInBytes() {
@@ -5488,6 +5890,7 @@ public class ConsensusPrefetchingQueue {
private static final class DeliveryBatchState {
private final List<Tablet> tablets = new ArrayList<>();
+ private final List<WriterBufferedEntry> entries = new ArrayList<>();
private final ByteArrayOutputStream replayPositions = new
ByteArrayOutputStream();
private long previousSearchIndex;
private long previousPhysicalTime;
@@ -5535,6 +5938,7 @@ public class ConsensusPrefetchingQueue {
endSearchIndex = entry.getSearchIndex();
}
tablets.addAll(entry.getTablets());
+ entries.add(entry);
DeliveryBatchRecord.writeDelta(replayPositions, entry.getSearchIndex() -
previousSearchIndex);
DeliveryBatchRecord.writeDelta(
replayPositions, entry.getPhysicalTime() - previousPhysicalTime);
@@ -5570,6 +5974,7 @@ public class ConsensusPrefetchingQueue {
private void reset() {
tablets.clear();
+ entries.clear();
replayPositions.reset();
previousSearchIndex = 0L;
previousPhysicalTime = 0L;
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 13568fd00be..ce33ee974bb 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;
@@ -128,6 +131,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;
@@ -454,6 +458,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;
}
@@ -801,6 +831,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 7a99e139733..783d520180d 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
@@ -66,7 +66,6 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
@@ -662,7 +661,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();
@@ -690,21 +689,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/ConsensusPrefetchingQueueDeliveryTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDeliveryTest.java
index 1f1a14535ca..e1ff4aaeb0b 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDeliveryTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDeliveryTest.java
@@ -69,6 +69,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
@@ -109,6 +110,7 @@ public class ConsensusPrefetchingQueueDeliveryTest {
private ConsensusLogToTabletConverter converter;
private ConsensusSubscriptionCommitManager commitManager;
private SubscriptionMemoryManager memoryManager;
+ private WriterSafeFrontierTracker writerSafeFrontierTracker;
private AtomicLong currentSearchIndex;
private long tabletBytes;
@@ -288,6 +290,43 @@ public class ConsensusPrefetchingQueueDeliveryTest {
assertTrue(queue.ack(CONSUMER, event.getCommitContext()));
}
+ @Test
+ public void testLingerBatchRemainsReplayableAfterAddingActiveWriter() throws
Exception {
+ config.setSubscriptionConsensusBatchMaxTabletCount(2);
+ config.setSubscriptionConsensusBatchMaxDelayInMs(60_000);
+ writeWal(2, true);
+ final ConsensusPrefetchingQueue queue = newQueue();
+ queue.setOrderMode(TopicConstant.ORDER_MODE_MULTI_WRITER_VALUE);
+ queue.setActiveWriterNodeIds(Collections.singleton(WRITER));
+ assertNull(queue.poll(CONSUMER));
+ assertTrue(pendingEntries(queue).offer(request(1, 1, WRITER)));
+ assertTrue(pendingEntries(queue).offer(request(2, 2, WRITER)));
+ queue.drivePrefetchOnce();
+ assertNull(queue.poll(CONSUMER));
+ assertEquals(tabletBytes * 2, queue.getRetainedTabletBytes());
+
+ writerSafeFrontierTracker.recordAppliedProgress(1002, WRITER, 2);
+ writerSafeFrontierTracker.observePendingWriterSafeTimeBarrier(2000,
WRITER, 2);
+ writerSafeFrontierTracker.observePendingWriterSafeTimeBarrier(2000, 8, 0);
+ queue.setActiveWriterNodeIds(Set.of(WRITER, 8));
+ config.setSubscriptionConsensusBatchMaxDelayInMs(0);
+ for (int round = 0; round < 20 && queue.getPrefetchedEventCount() == 0;
round++) {
+ queue.drivePrefetchOnce();
+ }
+ final SubscriptionEvent event = queue.poll(CONSUMER);
+ assertNotNull(event);
+ final SubscriptionPollResponse first = deliver(event);
+ assertEquals(Arrays.asList(1001L, 1002L), timestamps(first));
+ assertTrue(event.isPayloadDetached());
+ assertEquals(0L, queue.getRetainedTabletBytes());
+
+ assertTrue(queue.nack(CONSUMER, event.getCommitContext()));
+ assertSame(event, queue.poll(CONSUMER));
+ assertEquals(first.getPayload(), deliver(event).getPayload());
+ assertEquals(0L, queue.getRetainedTabletBytes());
+ assertTrue(queue.ack(CONSUMER, event.getCommitContext()));
+ }
+
@Test
public void testNackTimeoutDisconnectAndLateAckAfterReclamation() throws
Exception {
writeWal(1, true);
@@ -599,7 +638,8 @@ public class ConsensusPrefetchingQueueDeliveryTest {
private ConsensusPrefetchingQueue newQueue(final boolean
unboundedWalFixture) {
final IoTConsensusServerImpl server = mock(IoTConsensusServerImpl.class);
when(server.getConsensusReqReader()).thenReturn(walNode);
- when(server.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+ writerSafeFrontierTracker = new WriterSafeFrontierTracker();
+
when(server.getWriterSafeFrontierTracker()).thenReturn(writerSafeFrontierTracker);
final ConsensusPrefetchingQueue queue =
new ConsensusPrefetchingQueue(
"group",
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 ab5aff9b0d7..bd361edb08a 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(
@@ -171,36 +196,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..b34dd71b6f5
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWriterOrderingTest.java
@@ -0,0 +1,521 @@
+/*
+ * 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.lang.reflect.Method;
+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 testOrderedReplayResumesAfterSharedWalScanBudgetYields() throws
Exception {
+ newQueue(walNode(writeWal(1), 1), 1024 * 1024);
+ installSafeFrontiers(1);
+ final Method driveOrderedReplay =
+ ConsensusPrefetchingQueue.class.getDeclaredMethod(
+ "driveOrderedWriterReplay",
+ long.class,
+ int.class,
+ int.class,
+ long.class,
+ int.class,
+ ProgressWALIterator.ScanBudget.class);
+ driveOrderedReplay.setAccessible(true);
+ final ProgressWALIterator.ScanBudget scanBudget =
+ new ProgressWALIterator.ScanBudget(1, Long.MAX_VALUE);
+ final PrefetchRoundResult result =
+ (PrefetchRoundResult)
+ driveOrderedReplay.invoke(queue, 0L, 1024, 512, 16 * 1024 * 1024L,
0, scanBudget);
+
+ assertTrue(scanBudget.hasYielded());
+ assertEquals(PrefetchRoundResult.Type.RESCHEDULE_NOW, result.getType());
+ // A single physical entry remains buffered until the next writer request
establishes its end.
+ assertEquals(0, walReads.get());
+ assertEquals(0, conversions.get());
+ assertEquals(0L, queue.getRetainedTabletBytes());
+ assertNull(queue.poll("consumer"));
+
+ for (int round = 0; round < 20 && queue.getPrefetchedEventCount() == 0;
round++) {
+ queue.drivePrefetchOnce();
+ }
+ assertEvent(8, 1000);
+ consumeRemainingWriterSeven(1);
+ assertEquals(2, conversions.get());
+ }
+
+ @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);
+ }
+}
