This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 8e7990366 [server] Optimize RemoteLogFetcher with async prefetch for
recovery (#3132)
8e7990366 is described below
commit 8e799036656d5acf8a481a62c1b1e655f8edd25f
Author: Kaixuan Duan <[email protected]>
AuthorDate: Fri May 29 20:44:00 2026 +0800
[server] Optimize RemoteLogFetcher with async prefetch for recovery (#3132)
* [server] Optimize RemoteLogFetcher with async prefetch for recovery
* Add tests for RemoteLogFetcher prefetch fallback and cancellation
* Address some feedback
* Optimized the prefetch depth
* Address feedback
* Address feedback
* Optimize RemoteLogFetcher with async prefetch for KV recovery
* updated the docs
---
.../org/apache/fluss/config/ConfigOptions.java | 20 +
.../apache/fluss/server/kv/RemoteLogFetcher.java | 487 ++++++++++++++--
.../server/kv/snapshot/DefaultSnapshotContext.java | 24 +-
.../fluss/server/kv/snapshot/SnapshotContext.java | 17 +
.../org/apache/fluss/server/replica/Replica.java | 10 +-
.../fluss/server/kv/RemoteLogFetcherTest.java | 631 +++++++++++++++++++--
.../server/log/remote/TestingRemoteLogStorage.java | 105 ++++
website/docs/maintenance/configuration.md | 2 +
8 files changed, 1211 insertions(+), 85 deletions(-)
diff --git
a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
index b6154fa47..d0ede21a7 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -2032,6 +2032,26 @@ public class ConfigOptions {
.withDescription(
"The max fetch size for fetching log to apply to
kv during recovering kv.");
+ public static final ConfigOption<Integer>
KV_RECOVERY_REMOTE_LOG_PREFETCH_NUM =
+ key("kv.recover.remote-log.prefetch-num")
+ .intType()
+ .defaultValue(4)
+ .withDescription(
+ "The maximum number of remote log segments that
can be downloaded "
+ + "but not yet consumed during KV
recovery. A larger value "
+ + "overlaps more remote storage downloads
with consumption, "
+ + "at the cost of extra local disk usage.
Setting to 1 keeps "
+ + "the historical behavior (one-step
prefetch).");
+
+ public static final ConfigOption<Integer>
KV_RECOVERY_REMOTE_LOG_DOWNLOAD_THREADS =
+ key("kv.recover.remote-log.download-threads")
+ .intType()
+ .defaultValue(3)
+ .withDescription(
+ "The number of threads used to download remote log
segments during "
+ + "KV recovery. Should be less than or
equal to "
+ + "'kv.recover.remote-log.prefetch-num'.");
+
// ------------------------------------------------------------------------
// ConfigOptions for metrics
// ------------------------------------------------------------------------
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/kv/RemoteLogFetcher.java
b/fluss-server/src/main/java/org/apache/fluss/server/kv/RemoteLogFetcher.java
index 292a88726..b5943d838 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/kv/RemoteLogFetcher.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/kv/RemoteLogFetcher.java
@@ -25,6 +25,7 @@ import org.apache.fluss.record.LogRecordBatch;
import org.apache.fluss.remote.RemoteLogSegment;
import org.apache.fluss.server.log.remote.RemoteLogManager;
import org.apache.fluss.server.log.remote.RemoteLogStorage;
+import org.apache.fluss.utils.ExponentialBackoff;
import org.apache.fluss.utils.FlussPaths;
import org.apache.fluss.utils.IOUtils;
@@ -40,62 +41,96 @@ import java.io.InputStream;
import java.io.OutputStream;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.UUID;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.fluss.utils.FileUtils.deleteDirectoryQuietly;
/**
- * A utility class that fetches remote log segments and makes them available
as {@link
- * FileLogRecords} for KV recovery. It downloads remote log data into a local
temporary directory
- * using a UUID to avoid conflicts with other concurrent recovery operations.
+ * Fetches remote log segments as {@link FileLogRecords} for KV recovery,
downloading into a
+ * per-instance UUID temp directory. Must be closed after use (prefer
try-with-resources).
*
- * <p>The fetcher is {@link Closeable} and the caller must close it after use
to clean up the
- * temporary directory. It is recommended to use try-with-resources to ensure
proper resource
- * cleanup:
- *
- * <pre>{@code
- * try (RemoteLogFetcher fetcher = new RemoteLogFetcher(...)) {
- * for (LogRecordBatch batch : fetcher.fetch(startOffset,
localLogStartOffset)) {
- * // process batch
- * }
- * }
- * }</pre>
- *
- * <p><b>Note:</b> This class is NOT thread-safe. Each instance should be used
by a single thread
- * only.
+ * <p>Segments are prefetched in a bounded sliding window ({@code prefetchNum}
slots, {@code
+ * downloadThreads} concurrent downloads). As the consumer advances, consumed
slots are freed and
+ * back-filled, overlapping network I/O with local iteration.
*/
@NotThreadSafe
public class RemoteLogFetcher implements Closeable {
private static final Logger LOG =
LoggerFactory.getLogger(RemoteLogFetcher.class);
private static final String REMOTE_LOG_RECOVERY_DIR_PREFIX =
"remote-log-recovery-";
+ private static final long DOWNLOAD_RETRY_BACKOFF_INITIAL_MS = 100L;
+ private static final int DOWNLOAD_RETRY_BACKOFF_MULTIPLIER = 2;
+ private static final long DOWNLOAD_RETRY_BACKOFF_MAX_MS = 5_000L;
+ private static final double DOWNLOAD_RETRY_BACKOFF_JITTER = 0.25D;
+ @VisibleForTesting static final int DOWNLOAD_MAX_RETRIES = 5;
private final RemoteLogManager remoteLogManager;
private final TableBucket tableBucket;
private final Path tempDir;
+ private final ExecutorService downloadExecutor;
+ private final int prefetchNum;
/** Tracks the currently active iterator to ensure proper cleanup on
close. */
- private volatile RemoteLogBatchIterator activeIterator;
+ private RemoteLogBatchIterator activeIterator;
public RemoteLogFetcher(
- RemoteLogManager remoteLogManager, TableBucket tableBucket, File
logTabletDir) {
+ RemoteLogManager remoteLogManager,
+ TableBucket tableBucket,
+ File logTabletDir,
+ int prefetchNum,
+ int downloadThreads) {
this(
remoteLogManager,
tableBucket,
logTabletDir
.toPath()
.resolve("tmp")
- .resolve(REMOTE_LOG_RECOVERY_DIR_PREFIX +
UUID.randomUUID()));
+ .resolve(REMOTE_LOG_RECOVERY_DIR_PREFIX +
UUID.randomUUID()),
+ prefetchNum,
+ downloadThreads);
}
@VisibleForTesting
- RemoteLogFetcher(RemoteLogManager remoteLogManager, TableBucket
tableBucket, Path tempDir) {
+ RemoteLogFetcher(
+ RemoteLogManager remoteLogManager,
+ TableBucket tableBucket,
+ Path tempDir,
+ int prefetchNum,
+ int downloadThreads) {
this.remoteLogManager = remoteLogManager;
this.tableBucket = tableBucket;
this.tempDir = tempDir;
+ this.prefetchNum = Math.max(1, prefetchNum);
+ int threads = Math.max(1, Math.min(downloadThreads, this.prefetchNum));
+ AtomicInteger threadIndex = new AtomicInteger();
+ this.downloadExecutor =
+ Executors.newFixedThreadPool(
+ threads,
+ runnable -> {
+ Thread thread =
+ new Thread(
+ runnable,
+ "remote-log-fetcher-download-"
+ + tableBucket.getTableId()
+ + "-"
+ + tableBucket.getBucket()
+ + "-"
+ +
threadIndex.getAndIncrement());
+ thread.setDaemon(true);
+ return thread;
+ });
}
/**
@@ -136,31 +171,67 @@ public class RemoteLogFetcher implements Closeable {
}
LOG.info(
- "Found {} remote log segments for table bucket {} from offset
{} to localLogStartOffset {}",
+ "Found {} remote log segments for table bucket {} from offset
{} to localLogStartOffset {} "
+ + "(prefetchNum={}, downloadThreads={})",
segments.size(),
tableBucket,
startOffset,
- localLogStartOffset);
+ localLogStartOffset,
+ prefetchNum,
+ ((java.util.concurrent.ThreadPoolExecutor)
downloadExecutor).getCorePoolSize());
RemoteLogBatchIterator iterator =
new RemoteLogBatchIterator(segments, startOffset,
localLogStartOffset);
this.activeIterator = iterator;
+ // Kick off the initial prefetch window so downloads start before the
first advance().
+ iterator.fillPrefetchWindow();
return () -> iterator;
}
@Override
public void close() {
- // Close any active iterator to release file handles
- if (activeIterator != null) {
- activeIterator.close();
- activeIterator = null;
- }
- // Remove the entire "tmp" parent directory to clean up our
subdirectory as well
- // as any stale recovery directories left by a previous failed
recovery.
- Path tmpDir = tempDir.getParent();
- if (tmpDir != null && Files.exists(tmpDir)) {
- LOG.info("Cleaning up remote log recovery tmp dir: {}", tmpDir);
- deleteDirectoryQuietly(tmpDir.toFile());
+ try {
+ // Close any active iterator to release file handles
+ if (activeIterator != null) {
+ activeIterator.close();
+ activeIterator = null;
+ }
+ } finally {
+ downloadExecutor.shutdownNow();
+
+ boolean terminated = false;
+ try {
+ terminated =
+ downloadExecutor.awaitTermination(1,
java.util.concurrent.TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.warn(
+ "Interrupted while waiting for remote log fetcher
download executor termination for table bucket {}.",
+ tableBucket,
+ e);
+ }
+
+ if (!terminated) {
+ LOG.warn(
+ "Download executor did not terminate within 1 second
for table bucket {}. "
+ + "Proceeding with best-effort cleanup.",
+ tableBucket);
+ }
+
+ // Best-effort cleanup: always attempt to delete our own tempDir
regardless of
+ // whether the executor terminated. If a download is still in
progress the
+ // directory may not be fully removable, which is fine — we log
and move on.
+ if (Files.exists(tempDir)) {
+ LOG.info("Cleaning up remote log recovery temp dir: {}",
tempDir);
+ deleteDirectoryQuietly(tempDir.toFile());
+ if (Files.exists(tempDir)) {
+ LOG.warn(
+ "Failed to fully clean up remote log recovery temp
dir {} for table bucket {}. "
+ + "Some files may still be in use by an
ongoing download.",
+ tempDir,
+ tableBucket);
+ }
+ }
}
}
@@ -169,6 +240,24 @@ public class RemoteLogFetcher implements Closeable {
return tempDir;
}
+ @VisibleForTesting
+ int getPrefetchNum() {
+ return prefetchNum;
+ }
+
+ /**
+ * Returns an immutable snapshot of in-flight prefetch futures in
submission order. Must be
+ * called on the consumer thread while it is not advancing the iterator.
+ */
+ @VisibleForTesting
+ List<Future<File>> snapshotPrefetchFuturesForTest() {
+ RemoteLogBatchIterator iterator = this.activeIterator;
+ if (iterator == null) {
+ return Collections.emptyList();
+ }
+ return iterator.snapshotPrefetchFuturesForTest();
+ }
+
/**
* Downloads the log data of a remote log segment to a local temporary
file.
*
@@ -189,16 +278,85 @@ public class RemoteLogFetcher implements Closeable {
segment.remoteLogEndOffset(),
localFile);
+ boolean success = false;
try (InputStream inputStream = remoteLogStorage.fetchLogData(segment);
OutputStream outputStream =
Files.newOutputStream(localFile.toPath())) {
IOUtils.copyBytes(inputStream, outputStream, false);
+ success = true;
} catch (RemoteStorageException e) {
throw new IOException(
"Failed to download remote log segment: " +
segment.remoteLogSegmentId(), e);
+ } finally {
+ // Most InputStreams don't honor Thread.interrupt() mid-read, so
the copy may
+ // succeed despite interruption. Treat interrupt-after-success as
failure to
+ // avoid leaving stale files in tempDir.
+ if (!success || Thread.currentThread().isInterrupted()) {
+ try {
+ Files.deleteIfExists(localFile.toPath());
+ } catch (IOException cleanupException) {
+ LOG.warn(
+ "Failed to cleanup partial/interrupted local
segment file {} for segment {}.",
+ localFile,
+ segment.remoteLogSegmentId(),
+ cleanupException);
+ }
+ // File was deleted above; throw instead of returning a
dangling reference
+ // so the retry layer or consumer sees a clear failure.
+ if (success && Thread.currentThread().isInterrupted()) {
+ throw new IOException(
+ "Download completed but was interrupted for
segment "
+ + segment.remoteLogSegmentId());
+ }
+ }
}
return localFile;
}
+ private File downloadSegmentWithRetry(RemoteLogSegment segment) throws
IOException {
+ ExponentialBackoff backoff =
+ new ExponentialBackoff(
+ DOWNLOAD_RETRY_BACKOFF_INITIAL_MS,
+ DOWNLOAD_RETRY_BACKOFF_MULTIPLIER,
+ DOWNLOAD_RETRY_BACKOFF_MAX_MS,
+ DOWNLOAD_RETRY_BACKOFF_JITTER);
+
+ IOException lastException = null;
+ for (int attempt = 0; attempt <= DOWNLOAD_MAX_RETRIES; attempt++) {
+ try {
+ return downloadSegment(segment);
+ } catch (IOException e) {
+ lastException = e;
+ if (attempt == DOWNLOAD_MAX_RETRIES) {
+ break;
+ }
+
+ long retryDelayMs = backoff.backoff(attempt);
+ LOG.warn(
+ "Failed to download remote log segment {} on attempt
{}/{}. Retry after {} ms.",
+ segment.remoteLogSegmentId(),
+ attempt + 1,
+ DOWNLOAD_MAX_RETRIES + 1,
+ retryDelayMs,
+ e);
+
+ try {
+ Thread.sleep(retryDelayMs);
+ } catch (InterruptedException interruptedException) {
+ Thread.currentThread().interrupt();
+ throw new IOException(
+ "Interrupted while retrying remote log segment
download: "
+ + segment.remoteLogSegmentId(),
+ interruptedException);
+ }
+ }
+ }
+
+ throw new IOException(
+ "Failed to download remote log segment after retries: "
+ + segment.remoteLogSegmentId(),
+ lastException);
+ }
+
/**
* An iterator that lazily downloads remote log segments and iterates over
their batches in
* order. It respects the startOffset and localLogStartOffset boundaries,
yielding only batches
@@ -214,27 +372,74 @@ public class RemoteLogFetcher implements Closeable {
private int currentSegmentIndex = 0;
private FileLogRecords currentFileLogRecords;
private Iterator<LogRecordBatch> currentBatchIterator;
+
+ /**
+ * Ring of in-flight prefetch futures, indexed by {@code seq %
prefetchNum}. A non-null slot
+ * means "seq is in-flight or ready for consumption"; a null slot
means "free".
+ */
+ private final Future<File>[] prefetchSlots;
+
+ /**
+ * Parallel ring of the {@link RemoteLogSegment} each slot's future is
downloading. Kept
+ * alongside {@link #prefetchSlots} so the consumer can still verify
(via {@link
+ * #isSameSegmentId}) that the ring head matches the segment it is
about to request — a
+ * defensive check against test-injected mismatches and potential
future FIFO-ordering
+ * regressions.
+ */
+ private final RemoteLogSegment[] prefetchSegments;
+
+ /**
+ * Index of the next segment to be submitted for prefetching. Always
monotonically
+ * increasing.
+ */
+ private int nextPrefetchIndex = 0;
+
+ /**
+ * Index of the next segment to be consumed by the iterator. The
invariant {@code
+ * nextPrefetchIndex - nextConsumeIndex <= prefetchNum} enforces the
window size.
+ */
+ private int nextConsumeIndex = 0;
+
private LogRecordBatch nextBatch;
+ /** The local .log file currently opened as {@link
#currentFileLogRecords}. */
+ private File currentLocalFile;
+
private boolean finished = false;
- private volatile boolean closed = false;
+ private boolean closed = false;
RemoteLogBatchIterator(
List<RemoteLogSegment> segments, long startOffset, long
localLogStartOffset) {
this.segments = segments;
this.currentOffset = startOffset;
this.localLogStartOffset = localLogStartOffset;
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ Future<File>[] slots = (Future<File>[]) new Future[prefetchNum];
+ this.prefetchSlots = slots;
+ this.prefetchSegments = new RemoteLogSegment[prefetchNum];
+ }
+
+ /** Current number of segments in-flight or ready for consumption. */
+ private int prefetchWindowSize() {
+ return nextPrefetchIndex - nextConsumeIndex;
}
/** Closes this iterator and releases all held resources. */
public void close() {
if (!closed) {
closed = true;
+ finished = true; // Mark as finished to stop any ongoing
processing
+ nextBatch = null; // Clear any pending batch
+ drainPrefetchWindow();
closeCurrentFileLogRecords();
}
}
@Override
public boolean hasNext() {
+ // Return false immediately if closed
+ if (closed) {
+ return false;
+ }
// Lazily advance: only fetch next batch when needed. This ensures
the
// previously returned FileChannelLogRecordBatch has been fully
consumed
// by the caller before advance() potentially closes its
underlying file.
@@ -255,6 +460,12 @@ public class RemoteLogFetcher implements Closeable {
}
private void advance() {
+ // Early return if closed
+ if (closed) {
+ finished = true;
+ return;
+ }
+
nextBatch = null;
while (!finished) {
// try to get next batch from current iterator
@@ -297,12 +508,10 @@ public class RemoteLogFetcher implements Closeable {
}
try {
- // TODO optimize to async downloading the next segment
while processing the
- // current one. Otherwise, the recovery process may be
significantly slowed down
- // by the download time of remote segments, especially
when there are many
- // segments to fetch. trace by:
https://github.com/apache/fluss/issues/3091
- File localFile = downloadSegment(segment);
+ File localFile = fetchSegmentFile(segment);
+ currentLocalFile = localFile;
currentFileLogRecords = FileLogRecords.open(localFile,
false);
+ // fetchSegmentFile() already calls fillPrefetchWindow()
in its finally block.
int startPosition = 0;
// if this segment contains data before currentOffset,
find the right position
if (segment.remoteLogStartOffset() < currentOffset) {
@@ -324,7 +533,11 @@ public class RemoteLogFetcher implements Closeable {
currentBatchIterator =
currentFileLogRecords.batches().iterator();
}
} catch (Exception e) {
- // Ensure resources are cleaned up if an exception occurs
during segment loading
+ // Ensure resources are cleaned up if an exception occurs
during segment
+ // loading.
+ // After a failed segment, the recovery cannot continue,
so drop the entire
+ // window.
+ drainPrefetchWindow();
closeCurrentFileLogRecords();
throw new RuntimeException(
"Failed to fetch remote log segment: " +
segment.remoteLogSegmentId(),
@@ -333,12 +546,204 @@ public class RemoteLogFetcher implements Closeable {
}
}
+ /**
+ * Obtain the local file for {@code segment}, ideally reusing a
prefetched download. On
+ * success the ring head is consumed and a refill is triggered, so the
window stays full as
+ * long as more segments remain.
+ */
+ private File fetchSegmentFile(RemoteLogSegment segment) throws
IOException {
+ // advance() and fillPrefetchWindow() use identical skip rules in
FIFO order,
+ // so the ring head always matches the segment advance() is
requesting.
+ int headSlot = nextConsumeIndex % prefetchNum;
+ RemoteLogSegment headSegment = prefetchSegments[headSlot];
+ if (!isSameSegmentId(segment, headSegment)) {
+ // Ring head mismatch — drain the window and fail fast rather
than consume
+ // the wrong file.
+ drainPrefetchWindow();
+ throw new IOException(
+ "Prefetch ring head mismatch: requested "
+ + segment.remoteLogSegmentId()
+ + " but head is "
+ + (headSegment == null
+ ? "null"
+ : headSegment.remoteLogSegmentId()));
+ }
+ Future<File> headFuture = prefetchSlots[headSlot];
+ // Null the slot eagerly so the ring never keeps stale references.
+ prefetchSlots[headSlot] = null;
+ prefetchSegments[headSlot] = null;
+ nextConsumeIndex++;
+ try {
+ return headFuture.get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException(
+ "Interrupted while waiting for remote log segment
download: "
+ + segment.remoteLogSegmentId(),
+ e);
+ } catch (CancellationException e) {
+ LOG.warn(
+ "Prefetched segment {} was cancelled, fallback to sync
download.",
+ segment.remoteLogSegmentId(),
+ e);
+ return downloadSegmentWithRetry(segment);
+ } catch (ExecutionException e) {
+ LOG.warn(
+ "Prefetched segment {} failed even after async
retries.",
+ segment.remoteLogSegmentId(),
+ e.getCause());
+ throw new IOException(
+ "Failed to download remote log segment after async
retries: "
+ + segment.remoteLogSegmentId(),
+ e.getCause());
+ } finally {
+ // Window slot released — try to submit the next prefetch
right away.
+ fillPrefetchWindow();
+ }
+ }
+
+ private boolean isSameSegmentId(RemoteLogSegment left,
RemoteLogSegment right) {
+ return left != null
+ && right != null
+ &&
left.remoteLogSegmentId().equals(right.remoteLogSegmentId());
+ }
+
+ /**
+ * Submit as many download tasks as fit in the window. Non-blocking:
if the window is full
+ * we stop and wait for the consumer to advance.
+ */
+ private void fillPrefetchWindow() {
+ if (closed) {
+ return;
+ }
+
+ while (nextPrefetchIndex < segments.size()) {
+ RemoteLogSegment segment = segments.get(nextPrefetchIndex);
+
+ // segment entirely before currentOffset — skip and advance
index.
+ if (segment.remoteLogEndOffset() < currentOffset) {
+ nextPrefetchIndex++;
+ continue;
+ }
+ // segment starts at or beyond localLogStartOffset — nothing
more to prefetch.
+ if (segment.remoteLogStartOffset() >= localLogStartOffset) {
+ return;
+ }
+
+ // Window full — back off and wait for the consumer to advance.
+ if (prefetchWindowSize() >= prefetchNum) {
+ return;
+ }
+
+ final RemoteLogSegment target = segment;
+ Future<File> future;
+ try {
+ // submit(Callable) returns a FutureTask whose
cancel(true) interrupts the
+ // worker thread, unlike CompletableFuture which only
flips state.
+ future =
+ downloadExecutor.submit(
+ (Callable<File>) () ->
downloadSegmentWithRetry(target));
+ } catch (Throwable submitError) {
+ // The executor rejected the task (e.g. we're shutting
down); stop refilling.
+ // No slot has been claimed yet, so there is nothing to
roll back.
+ LOG.debug(
+ "Failed to submit prefetch for segment {}
(executor likely shutting down).",
+ target.remoteLogSegmentId(),
+ submitError);
+ return;
+ }
+
+ int slot = nextPrefetchIndex % prefetchNum;
+ prefetchSlots[slot] = future;
+ prefetchSegments[slot] = target;
+ nextPrefetchIndex++;
+ LOG.debug(
+ "Prefetching remote log segment {} for bucket {}
(window size={}, free slots={}).",
+ target.remoteLogSegmentId(),
+ tableBucket,
+ prefetchWindowSize(),
+ prefetchNum - prefetchWindowSize());
+ }
+ }
+
+ /** Cancel or clean up all entries in the window, leaving it empty. */
+ private void drainPrefetchWindow() {
+ while (nextConsumeIndex < nextPrefetchIndex) {
+ int slot = nextConsumeIndex % prefetchNum;
+ Future<File> future = prefetchSlots[slot];
+ prefetchSlots[slot] = null;
+ prefetchSegments[slot] = null;
+ nextConsumeIndex++;
+ if (future.isDone()) {
+ if (!cleanupCompletedFuture(future)) {
+ return;
+ }
+ } else {
+ if (!future.cancel(true) &&
!cleanupCompletedFuture(future)) {
+ return;
+ }
+ }
+ }
+ }
+
+ /**
+ * Retrieve and delete the file from a completed future. Returns
{@code false} if
+ * interrupted (caller should bail out).
+ */
+ private boolean cleanupCompletedFuture(Future<File> future) {
+ try {
+ // isDone() is true so get() will not block; it returns either
the
+ // successfully downloaded file, or throws
Cancellation/Execution.
+ cleanupUnusedPrefetchedFile(future.get());
+ } catch (CancellationException | ExecutionException ignored) {
+ // no local file to clean up
+ } catch (InterruptedException e) {
+ // Interrupt came from elsewhere; restore flag and bail out.
close()'s
+ // shutdownNow path will reclaim remaining slots.
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ return true;
+ }
+
+ private void cleanupUnusedPrefetchedFile(File prefetchedFile) {
+ if (prefetchedFile == null) {
+ return;
+ }
+
+ try {
+ Files.deleteIfExists(prefetchedFile.toPath());
+ } catch (IOException cleanupException) {
+ LOG.warn(
+ "Failed to cleanup unused prefetched segment file {}
for table bucket {}.",
+ prefetchedFile,
+ tableBucket,
+ cleanupException);
+ }
+ }
+
private void closeCurrentFileLogRecords() {
if (currentFileLogRecords != null) {
IOUtils.closeQuietly(currentFileLogRecords, "FileLogRecords");
currentFileLogRecords = null;
currentBatchIterator = null;
}
+ if (currentLocalFile != null) {
+ try {
+ Files.deleteIfExists(currentLocalFile.toPath());
+ } catch (IOException e) {
+ LOG.warn("Failed to delete consumed segment file {}",
currentLocalFile, e);
+ }
+ currentLocalFile = null;
+ }
+ }
+
+ List<Future<File>> snapshotPrefetchFuturesForTest() {
+ List<Future<File>> snapshot = new
ArrayList<>(prefetchWindowSize());
+ for (int i = nextConsumeIndex; i < nextPrefetchIndex; i++) {
+ snapshot.add(prefetchSlots[i % prefetchNum]);
+ }
+ return Collections.unmodifiableList(snapshot);
}
}
}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/DefaultSnapshotContext.java
b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/DefaultSnapshotContext.java
index 9bec8977c..492d8aca4 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/DefaultSnapshotContext.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/DefaultSnapshotContext.java
@@ -57,6 +57,10 @@ public class DefaultSnapshotContext implements
SnapshotContext, ServerReconfigur
private final int maxFetchLogSizeInRecoverKv;
+ private final int remoteLogPrefetchNumInRecoverKv;
+
+ private final int remoteLogDownloadThreadsInRecoverKv;
+
private final FsPath remoteKvDir;
private DefaultSnapshotContext(
@@ -70,7 +74,9 @@ public class DefaultSnapshotContext implements
SnapshotContext, ServerReconfigur
int writeBufferSizeInBytes,
FsPath remoteKvDir,
CompletedSnapshotHandleStore completedSnapshotHandleStore,
- int maxFetchLogSizeInRecoverKv) {
+ int maxFetchLogSizeInRecoverKv,
+ int remoteLogPrefetchNumInRecoverKv,
+ int remoteLogDownloadThreadsInRecoverKv) {
this.zooKeeperClient = zooKeeperClient;
this.completedKvSnapshotCommitter = completedKvSnapshotCommitter;
this.snapshotScheduler = snapshotScheduler;
@@ -83,6 +89,8 @@ public class DefaultSnapshotContext implements
SnapshotContext, ServerReconfigur
this.completedSnapshotHandleStore = completedSnapshotHandleStore;
this.maxFetchLogSizeInRecoverKv = maxFetchLogSizeInRecoverKv;
+ this.remoteLogPrefetchNumInRecoverKv = remoteLogPrefetchNumInRecoverKv;
+ this.remoteLogDownloadThreadsInRecoverKv =
remoteLogDownloadThreadsInRecoverKv;
}
public static DefaultSnapshotContext create(
@@ -101,7 +109,9 @@ public class DefaultSnapshotContext implements
SnapshotContext, ServerReconfigur
(int)
conf.get(ConfigOptions.REMOTE_FS_WRITE_BUFFER_SIZE).getBytes(),
FlussPaths.remoteKvDir(conf),
new ZooKeeperCompletedSnapshotHandleStore(zkClient),
- (int)
conf.get(ConfigOptions.KV_RECOVER_LOG_RECORD_BATCH_MAX_SIZE).getBytes());
+ (int)
conf.get(ConfigOptions.KV_RECOVER_LOG_RECORD_BATCH_MAX_SIZE).getBytes(),
+ conf.get(ConfigOptions.KV_RECOVERY_REMOTE_LOG_PREFETCH_NUM),
+
conf.get(ConfigOptions.KV_RECOVERY_REMOTE_LOG_DOWNLOAD_THREADS));
}
public ZooKeeperClient getZooKeeperClient() {
@@ -163,6 +173,16 @@ public class DefaultSnapshotContext implements
SnapshotContext, ServerReconfigur
return maxFetchLogSizeInRecoverKv;
}
+ @Override
+ public int remoteLogPrefetchNumInRecoverKv() {
+ return remoteLogPrefetchNumInRecoverKv;
+ }
+
+ @Override
+ public int remoteLogDownloadThreadsInRecoverKv() {
+ return remoteLogDownloadThreadsInRecoverKv;
+ }
+
@Override
public void handleSnapshotBroken(CompletedSnapshot snapshot) throws
Exception {
completedSnapshotHandleStore.remove(snapshot.getTableBucket(),
snapshot.getSnapshotID());
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotContext.java
b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotContext.java
index 8f528bc47..18c32d531 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotContext.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotContext.java
@@ -17,6 +17,7 @@
package org.apache.fluss.server.kv.snapshot;
+import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.fs.FsPath;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.server.zk.ZooKeeperClient;
@@ -82,4 +83,20 @@ public interface SnapshotContext {
* log during recovering.
*/
int maxFetchLogSizeInRecoverKv();
+
+ /**
+ * Get the maximum number of remote log segments that can be downloaded
but not yet consumed
+ * during KV recovery. See {@code kv.recover.remote-log.prefetch-num}.
+ */
+ default int remoteLogPrefetchNumInRecoverKv() {
+ return
ConfigOptions.KV_RECOVERY_REMOTE_LOG_PREFETCH_NUM.defaultValue();
+ }
+
+ /**
+ * Get the number of threads used to download remote log segments during
KV recovery. See {@code
+ * kv.recover.remote-log.download-threads}.
+ */
+ default int remoteLogDownloadThreadsInRecoverKv() {
+ return
ConfigOptions.KV_RECOVERY_REMOTE_LOG_DOWNLOAD_THREADS.defaultValue();
+ }
}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
index 1f8b893ab..e11dd69c1 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
@@ -899,10 +899,14 @@ public final class Replica {
snapshotContext.getZooKeeperClient(),
snapshotContext.maxFetchLogSizeInRecoverKv());
- // Always create RemoteLogFetcher; the temp directory is lazily
created only
- // when fetch() is actually called, so this is lightweight.
+ // Temp directory is created lazily on first fetch().
RemoteLogFetcher remoteLogFetcher =
- new RemoteLogFetcher(remoteLogManager, tableBucket,
logTablet.getLogDir());
+ new RemoteLogFetcher(
+ remoteLogManager,
+ tableBucket,
+ logTablet.getLogDir(),
+ snapshotContext.remoteLogPrefetchNumInRecoverKv(),
+
snapshotContext.remoteLogDownloadThreadsInRecoverKv());
try {
KvRecoverHelper kvRecoverHelper =
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/kv/RemoteLogFetcherTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/kv/RemoteLogFetcherTest.java
index fcb4e1cad..d8bd5e9d9 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/kv/RemoteLogFetcherTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/kv/RemoteLogFetcherTest.java
@@ -28,10 +28,19 @@ import org.apache.fluss.server.replica.Replica;
import org.junit.jupiter.api.Test;
import java.io.File;
+import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
+import java.util.Iterator;
import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.fluss.record.TestData.DATA1_TABLE_ID;
import static org.assertj.core.api.Assertions.assertThat;
@@ -40,6 +49,26 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for {@link RemoteLogFetcher}. */
class RemoteLogFetcherTest extends RemoteLogTestBase {
+ /** Default knobs for the tests: prefetch depth 2, 2 download threads. */
+ private static final int DEFAULT_TEST_PREFETCH_NUM = 2;
+
+ private static final int DEFAULT_TEST_DOWNLOAD_THREADS = 2;
+
+ private RemoteLogFetcher newFetcher(TableBucket tb, File logTabletDir) {
+ return new RemoteLogFetcher(
+ remoteLogManager,
+ tb,
+ logTabletDir,
+ DEFAULT_TEST_PREFETCH_NUM,
+ DEFAULT_TEST_DOWNLOAD_THREADS);
+ }
+
+ private RemoteLogFetcher newFetcher(
+ TableBucket tb, File logTabletDir, int prefetchNum, int
downloadThreads) {
+ return new RemoteLogFetcher(
+ remoteLogManager, tb, logTabletDir, prefetchNum,
downloadThreads);
+ }
+
@Test
void testBasicFetch() throws Exception {
TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
@@ -59,7 +88,7 @@ class RemoteLogFetcherTest extends RemoteLogTestBase {
long remoteEndOffset = segments.get(segments.size() -
1).remoteLogEndOffset();
File logTabletDir = logTablet.getLogDir();
- try (RemoteLogFetcher fetcher = new RemoteLogFetcher(remoteLogManager,
tb, logTabletDir)) {
+ try (RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir)) {
// Fetch all remote log batches from offset 0 to remoteEndOffset.
Iterable<LogRecordBatch> batches = fetcher.fetch(0,
remoteEndOffset);
List<LogRecordBatch> batchList = new ArrayList<>();
@@ -97,7 +126,7 @@ class RemoteLogFetcherTest extends RemoteLogTestBase {
long remoteEndOffset = segments.get(segments.size() -
1).remoteLogEndOffset();
File logTabletDir = logTablet.getLogDir();
- try (RemoteLogFetcher fetcher = new RemoteLogFetcher(remoteLogManager,
tb, logTabletDir)) {
+ try (RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir)) {
Iterable<LogRecordBatch> batches = fetcher.fetch(startOffset,
remoteEndOffset);
List<LogRecordBatch> batchList = new ArrayList<>();
for (LogRecordBatch batch : batches) {
@@ -129,7 +158,7 @@ class RemoteLogFetcherTest extends RemoteLogTestBase {
long localLogStartOffset = segments.get(1).remoteLogStartOffset();
File logTabletDir = logTablet.getLogDir();
- try (RemoteLogFetcher fetcher = new RemoteLogFetcher(remoteLogManager,
tb, logTabletDir)) {
+ try (RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir)) {
Iterable<LogRecordBatch> batches = fetcher.fetch(0,
localLogStartOffset);
List<LogRecordBatch> batchList = new ArrayList<>();
for (LogRecordBatch batch : batches) {
@@ -153,7 +182,7 @@ class RemoteLogFetcherTest extends RemoteLogTestBase {
Replica replica = replicaManager.getReplicaOrException(tb);
File logTabletDir = replica.getLogTablet().getLogDir();
- try (RemoteLogFetcher fetcher = new RemoteLogFetcher(remoteLogManager,
tb, logTabletDir)) {
+ try (RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir)) {
assertThatThrownBy(() -> fetcher.fetch(0, 100))
.isInstanceOf(RemoteStorageException.class)
.hasMessageContaining("No remote log segments found");
@@ -161,7 +190,7 @@ class RemoteLogFetcherTest extends RemoteLogTestBase {
}
@Test
- void testStaleDirectoriesFromUncleanShutdownAreCleanedUp() throws
Exception {
+ void testFetcherCleansUpItsOwnTempDirOnClose() throws Exception {
TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
makeLogTableAsLeader(tb, false);
Replica replica = replicaManager.getReplicaOrException(tb);
@@ -170,23 +199,23 @@ class RemoteLogFetcherTest extends RemoteLogTestBase {
Path tmpDir = logTabletDir.toPath().resolve("tmp");
Files.createDirectories(tmpDir);
- // Simulate stale directories left behind by a previous unclean
shutdown.
Path staleDir1 =
Files.createDirectories(tmpDir.resolve("remote-log-recovery-stale-1"));
Path staleDir2 =
Files.createDirectories(tmpDir.resolve("remote-log-recovery-stale-2"));
- // Create a file inside one stale dir to simulate a partially
downloaded segment.
Files.createFile(staleDir1.resolve("00000000000000000000.log"));
assertThat(Files.exists(staleDir1)).isTrue();
assertThat(Files.exists(staleDir2)).isTrue();
- // Simulate next startup: creating and closing a new RemoteLogFetcher
should
- // clean up the entire tmp directory including all stale recovery
directories.
- RemoteLogFetcher fetcher = new RemoteLogFetcher(remoteLogManager, tb,
logTabletDir);
+ // Only the fetcher's own UUID dir should be removed, stale dirs left
intact.
+ RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir);
+ Path fetcherTempDir = fetcher.getTempDir();
+ Files.createDirectories(fetcherTempDir);
fetcher.close();
- assertThat(Files.exists(staleDir1)).isFalse();
- assertThat(Files.exists(staleDir2)).isFalse();
- assertThat(Files.exists(tmpDir)).isFalse();
+ assertThat(Files.exists(fetcherTempDir)).isFalse();
+ assertThat(Files.exists(staleDir1)).isTrue();
+ assertThat(Files.exists(staleDir2)).isTrue();
+ assertThat(Files.exists(tmpDir)).isTrue();
}
@Test
@@ -204,23 +233,18 @@ class RemoteLogFetcherTest extends RemoteLogTestBase {
long remoteEndOffset = segments.get(segments.size() -
1).remoteLogEndOffset();
File logTabletDir = logTablet.getLogDir();
- RemoteLogFetcher fetcher = new RemoteLogFetcher(remoteLogManager, tb,
logTabletDir);
+ RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir);
Path tempDir = fetcher.getTempDir();
- // Not yet created before fetch.
assertThat(Files.exists(tempDir)).isFalse();
- // Fetch triggers lazy directory creation.
for (LogRecordBatch batch : fetcher.fetch(0, remoteEndOffset)) {
assertThat(Files.exists(tempDir)).isTrue();
}
assertThat(Files.exists(tempDir)).isTrue();
- // Close and verify cleanup.
fetcher.close();
assertThat(Files.exists(tempDir)).isFalse();
- // The parent "tmp" directory should also be removed when empty.
- assertThat(Files.exists(tempDir.getParent())).isFalse();
}
@Test
@@ -229,30 +253,26 @@ class RemoteLogFetcherTest extends RemoteLogTestBase {
makeLogTableAsLeader(tb, false);
Replica replica = replicaManager.getReplicaOrException(tb);
LogTablet logTablet = replica.getLogTablet();
- // Add more segments to test multi-segment iteration.
addMultiSegmentsToLogTablet(logTablet, 8);
remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
List<RemoteLogSegment> segments =
remoteLogManager.relevantRemoteLogSegments(tb, 0L);
- // We expect at least 7 remote segments (8 total segments, last one is
active).
assertThat(segments).hasSizeGreaterThanOrEqualTo(4);
long remoteEndOffset = segments.get(segments.size() -
1).remoteLogEndOffset();
File logTabletDir = logTablet.getLogDir();
- try (RemoteLogFetcher fetcher = new RemoteLogFetcher(remoteLogManager,
tb, logTabletDir)) {
+ try (RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir)) {
Iterable<LogRecordBatch> batches = fetcher.fetch(0,
remoteEndOffset);
long prevNextOffset = 0;
int batchCount = 0;
for (LogRecordBatch batch : batches) {
- // Verify strict monotonic ordering of offsets.
assertThat(batch.baseLogOffset()).isGreaterThanOrEqualTo(prevNextOffset);
prevNextOffset = batch.nextLogOffset();
batchCount++;
}
- // Verify we got a reasonable number of batches across all
segments.
assertThat(batchCount).isGreaterThan(10);
}
}
@@ -270,15 +290,11 @@ class RemoteLogFetcherTest extends RemoteLogTestBase {
List<RemoteLogSegment> segments =
remoteLogManager.relevantRemoteLogSegments(tb, 0L);
assertThat(segments).isNotEmpty();
- // Use the same offset for both startOffset and localLogStartOffset,
- // creating an empty range [x, x). No batches should be returned.
- // This guards against the >= comparison in advance() being
accidentally
- // changed to >, which would leak batches and cause duplicate records
- // during KV recovery.
+ // Empty range [x, x) — guards against advance() >= being changed to >.
long sameOffset = segments.get(0).remoteLogStartOffset();
File logTabletDir = logTablet.getLogDir();
- try (RemoteLogFetcher fetcher = new RemoteLogFetcher(remoteLogManager,
tb, logTabletDir)) {
+ try (RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir)) {
Iterable<LogRecordBatch> batches = fetcher.fetch(sameOffset,
sameOffset);
List<LogRecordBatch> batchList = new ArrayList<>();
for (LogRecordBatch batch : batches) {
@@ -303,22 +319,559 @@ class RemoteLogFetcherTest extends RemoteLogTestBase {
long remoteEndOffset = segments.get(segments.size() -
1).remoteLogEndOffset();
File logTabletDir = logTablet.getLogDir();
- try (RemoteLogFetcher fetcher = new RemoteLogFetcher(remoteLogManager,
tb, logTabletDir)) {
+ try (RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir)) {
Path tempDir = fetcher.getTempDir();
// Consume all batches to trigger downloading.
+ int batchCount = 0;
+ for (LogRecordBatch batch : fetcher.fetch(0, remoteEndOffset)) {
+ batchCount++;
+ // Check mid-iteration; files are deleted once the segment is
consumed.
+ if (batchCount == 1) {
+ File[] downloadedFiles = tempDir.toFile().listFiles();
+ assertThat(downloadedFiles).isNotNull();
+ assertThat(downloadedFiles.length).isGreaterThan(0);
+ for (File file : downloadedFiles) {
+ assertThat(file.getName()).endsWith(".log");
+ }
+ }
+ }
+ assertThat(batchCount).isGreaterThan(0);
+ assertThat(countLogFiles(tempDir)).isZero();
+ }
+ }
+
+ @Test
+ void testPrefetchFailurePropagatesAfterAsyncRetriesExhausted() throws
Exception {
+ TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
+ makeLogTableAsLeader(tb, false);
+ Replica replica = replicaManager.getReplicaOrException(tb);
+ LogTablet logTablet = replica.getLogTablet();
+ addMultiSegmentsToLogTablet(logTablet, 5);
+
+ remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+
+ List<RemoteLogSegment> segments =
remoteLogManager.relevantRemoteLogSegments(tb, 0L);
+ assertThat(segments).hasSizeGreaterThanOrEqualTo(2);
+ long remoteEndOffset = segments.get(segments.size() -
1).remoteLogEndOffset();
+
+ // Exhaust retries for segment #1 so the async worker raises
ExecutionException.
+ int prefetchAttempts = RemoteLogFetcher.DOWNLOAD_MAX_RETRIES + 1;
+ UUID failingSegmentId = segments.get(1).remoteLogSegmentId();
+ remoteLogStorage.fetchLogDataFailureBudget.put(
+ failingSegmentId, new AtomicInteger(prefetchAttempts));
+
+ File logTabletDir = logTablet.getLogDir();
+ try (RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir)) {
+ Iterator<LogRecordBatch> iterator = fetcher.fetch(0,
remoteEndOffset).iterator();
+
+ assertThatThrownBy(
+ () -> {
+ while (iterator.hasNext()) {
+ iterator.next();
+ }
+ })
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to fetch remote log
segment");
+
+ // Budget fully drained — proves prefetch exhausted retries.
+
assertThat(remoteLogStorage.fetchLogDataFailureBudget.get(failingSegmentId).get())
+ .as("prefetch retries should be fully exhausted")
+ .isZero();
+ } finally {
+ remoteLogStorage.fetchLogDataFailureBudget.clear();
+ }
+ }
+
+ @Test
+ void testCloseCancelsPendingPrefetchFuture() throws Exception {
+ TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
+ makeLogTableAsLeader(tb, false);
+ Replica replica = replicaManager.getReplicaOrException(tb);
+ LogTablet logTablet = replica.getLogTablet();
+ addMultiSegmentsToLogTablet(logTablet, 5);
+
+ remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+
+ List<RemoteLogSegment> segments =
remoteLogManager.relevantRemoteLogSegments(tb, 0L);
+ assertThat(segments).isNotEmpty();
+ long remoteEndOffset = segments.get(segments.size() -
1).remoteLogEndOffset();
+
+ // Block workers in fetchLogData; the only way out is
Thread.interrupt().
+ int expectedInFlight = 2;
+ CountDownLatch entered = new CountDownLatch(expectedInFlight);
+ CountDownLatch release = new CountDownLatch(1);
+ remoteLogStorage.fetchLogDataBarrier =
+ () -> {
+ entered.countDown();
+ release.await();
+ };
+
+ File logTabletDir = logTablet.getLogDir();
+ RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir, 2, 2);
+ try {
+ // fetch() is lazy — kicks off prefetch workers that will block on
`release`.
+ fetcher.fetch(0, remoteEndOffset);
+
+ assertThat(entered.await(30, TimeUnit.SECONDS))
+ .as("prefetch workers should have entered fetchLogData")
+ .isTrue();
+
+ // Snapshot pending futures BEFORE close so we can assert on them
afterwards.
+ List<Future<File>> inflight =
fetcher.snapshotPrefetchFuturesForTest();
+ assertThat(inflight).isNotEmpty();
+
+ fetcher.close();
+
+ // Every captured pending future must end up done and cancelled or
exceptional
+ // (an interrupted worker surfaces as RemoteStorageException →
ExecutionException).
+ for (Future<File> f : inflight) {
+ assertThat(f.isDone()).isTrue();
+ boolean exceptional = false;
+ try {
+ f.get();
+ } catch (CancellationException | ExecutionException expected) {
+ exceptional = true;
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ }
+ assertThat(f.isCancelled() || exceptional).isTrue();
+ }
+ } finally {
+ remoteLogStorage.fetchLogDataBarrier = null;
+ release.countDown();
+ fetcher.close();
+ }
+ }
+
+ @Test
+ void testSecondFetchClosesPreviousIteratorAndOldIteratorStops() throws
Exception {
+ TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
+ makeLogTableAsLeader(tb, false);
+ Replica replica = replicaManager.getReplicaOrException(tb);
+ LogTablet logTablet = replica.getLogTablet();
+ addMultiSegmentsToLogTablet(logTablet, 5);
+
+ remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+
+ List<RemoteLogSegment> segments =
remoteLogManager.relevantRemoteLogSegments(tb, 0L);
+ assertThat(segments).hasSizeGreaterThanOrEqualTo(2);
+ long remoteEndOffset = segments.get(segments.size() -
1).remoteLogEndOffset();
+
+ File logTabletDir = logTablet.getLogDir();
+ try (RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir)) {
+ Iterable<LogRecordBatch> firstIterable = fetcher.fetch(0,
remoteEndOffset);
+ Iterator<LogRecordBatch> firstIterator = firstIterable.iterator();
+
+ assertThat(firstIterator.hasNext()).isTrue();
+ firstIterator.next();
+
+ // Trigger a second fetch. This should close the previous active
iterator.
+ Iterable<LogRecordBatch> secondIterable = fetcher.fetch(0,
remoteEndOffset);
+ Iterator<LogRecordBatch> secondIterator =
secondIterable.iterator();
+
+ assertThat(firstIterator.hasNext()).isFalse();
+ assertThat(secondIterator.hasNext()).isTrue();
+ }
+ }
+
+ @Test
+ void testSecondFetchAfterRealAsyncPrefetchDoesNotLetOldIteratorContinue()
throws Exception {
+ TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
+ makeLogTableAsLeader(tb, false);
+ Replica replica = replicaManager.getReplicaOrException(tb);
+ LogTablet logTablet = replica.getLogTablet();
+ addMultiSegmentsToLogTablet(logTablet, 8);
+
+ remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+
+ List<RemoteLogSegment> segments =
remoteLogManager.relevantRemoteLogSegments(tb, 0L);
+ assertThat(segments).hasSizeGreaterThanOrEqualTo(2);
+ long remoteEndOffset = segments.get(segments.size() -
1).remoteLogEndOffset();
+
+ File logTabletDir = logTablet.getLogDir();
+ try (RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir)) {
+ Path tempDir = fetcher.getTempDir();
+
+ Iterator<LogRecordBatch> firstIterator = fetcher.fetch(0,
remoteEndOffset).iterator();
+
+ assertThat(firstIterator.hasNext()).isTrue();
+ firstIterator.next();
+
+ waitUntilLogFileCountAtLeast(tempDir, 2);
+
+ long fileCountBeforeSecondFetch = countLogFiles(tempDir);
+ assertThat(fileCountBeforeSecondFetch).isGreaterThanOrEqualTo(2);
+
+ Iterator<LogRecordBatch> secondIterator = fetcher.fetch(0,
remoteEndOffset).iterator();
+
+ assertThat(firstIterator.hasNext()).isFalse();
+ assertThat(secondIterator.hasNext()).isTrue();
+ assertThat(Files.exists(tempDir)).isTrue();
+ }
+ }
+
+ //
======================================================================================
+ // New tests specific to the configurable prefetch window / multi-threaded
download
+ //
======================================================================================
+
+ @Test
+ void testPrefetchDepthGreaterThanOnePrefetchesMultipleSegments() throws
Exception {
+ TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
+ makeLogTableAsLeader(tb, false);
+ Replica replica = replicaManager.getReplicaOrException(tb);
+ LogTablet logTablet = replica.getLogTablet();
+ addMultiSegmentsToLogTablet(logTablet, 8);
+
+ remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+
+ List<RemoteLogSegment> segments =
remoteLogManager.relevantRemoteLogSegments(tb, 0L);
+ assertThat(segments).hasSizeGreaterThanOrEqualTo(4);
+ long remoteEndOffset = segments.get(segments.size() -
1).remoteLogEndOffset();
+
+ File logTabletDir = logTablet.getLogDir();
+ try (RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir, 3, 3)) {
+ Path tempDir = fetcher.getTempDir();
+ Iterator<LogRecordBatch> iterator = fetcher.fetch(0,
remoteEndOffset).iterator();
+ assertThat(iterator.hasNext()).isTrue();
+ iterator.next();
+
+ // Window refills after first segment is opened.
+ waitUntilLogFileCountAtLeast(tempDir, 3);
+ assertThat(countLogFiles(tempDir)).isGreaterThanOrEqualTo(3);
+ }
+ }
+
+ @Test
+ void testPrefetchWindowRefillsAfterSegmentConsumed() throws Exception {
+ TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
+ makeLogTableAsLeader(tb, false);
+ Replica replica = replicaManager.getReplicaOrException(tb);
+ LogTablet logTablet = replica.getLogTablet();
+ addMultiSegmentsToLogTablet(logTablet, 8);
+
+ remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+
+ List<RemoteLogSegment> segments =
remoteLogManager.relevantRemoteLogSegments(tb, 0L);
+ assertThat(segments).hasSizeGreaterThanOrEqualTo(4);
+ long remoteEndOffset = segments.get(segments.size() -
1).remoteLogEndOffset();
+
+ File logTabletDir = logTablet.getLogDir();
+ // Small window forces repeated refills.
+ try (RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir, 2, 2)) {
+ Iterable<LogRecordBatch> batches = fetcher.fetch(0,
remoteEndOffset);
+ int batchCount = 0;
+ long prevNextOffset = 0;
+ for (LogRecordBatch batch : batches) {
+
assertThat(batch.baseLogOffset()).isGreaterThanOrEqualTo(prevNextOffset);
+ prevNextOffset = batch.nextLogOffset();
+ batchCount++;
+ }
+ // With 4+ segments and 10 batches each we expect clearly more
than 2*windowSize.
+ assertThat(batchCount).isGreaterThan(10);
+ }
+ }
+
+ @Test
+ void testPrefetchDepthOfOnePreservesLegacyBehavior() throws Exception {
+ TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
+ makeLogTableAsLeader(tb, false);
+ Replica replica = replicaManager.getReplicaOrException(tb);
+ LogTablet logTablet = replica.getLogTablet();
+ addMultiSegmentsToLogTablet(logTablet, 5);
+
+ remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+
+ List<RemoteLogSegment> segments =
remoteLogManager.relevantRemoteLogSegments(tb, 0L);
+ assertThat(segments).isNotEmpty();
+ long remoteEndOffset = segments.get(segments.size() -
1).remoteLogEndOffset();
+
+ File logTabletDir = logTablet.getLogDir();
+ try (RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir, 1, 1)) {
+ assertThat(fetcher.getPrefetchNum()).isEqualTo(1);
+
+ Iterable<LogRecordBatch> batches = fetcher.fetch(0,
remoteEndOffset);
+ int batchCount = 0;
+ long prevNextOffset = 0;
+ for (LogRecordBatch batch : batches) {
+
assertThat(batch.baseLogOffset()).isGreaterThanOrEqualTo(prevNextOffset);
+ prevNextOffset = batch.nextLogOffset();
+ batchCount++;
+ }
+ assertThat(batchCount).isGreaterThan(0);
+ }
+ }
+
+ @Test
+ void testCloseCancelsAllPendingPrefetchFutures() throws Exception {
+ TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
+ makeLogTableAsLeader(tb, false);
+ Replica replica = replicaManager.getReplicaOrException(tb);
+ LogTablet logTablet = replica.getLogTablet();
+ addMultiSegmentsToLogTablet(logTablet, 5);
+
+ remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+
+ List<RemoteLogSegment> segments =
remoteLogManager.relevantRemoteLogSegments(tb, 0L);
+ assertThat(segments).isNotEmpty();
+ long remoteEndOffset = segments.get(segments.size() -
1).remoteLogEndOffset();
+
+ // Block real prefetch workers so multiple futures are pending in
parallel. Focus
+ // here (vs testCloseCancelsPendingPrefetchFuture) is the *count*
contract:
+ // snapshot() returns all in-flight futures and close() cancels every
single one.
+ int expectedInFlight = 2;
+ CountDownLatch entered = new CountDownLatch(expectedInFlight);
+ CountDownLatch release = new CountDownLatch(1);
+ remoteLogStorage.fetchLogDataBarrier =
+ () -> {
+ entered.countDown();
+ release.await();
+ };
+
+ File logTabletDir = logTablet.getLogDir();
+ RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir, 2, 2);
+ try {
+ // fetch() is lazy — kicks off prefetch workers that will all
block on `release`.
+ fetcher.fetch(0, remoteEndOffset);
+
+ assertThat(entered.await(30, TimeUnit.SECONDS))
+ .as("prefetch workers should have entered fetchLogData")
+ .isTrue();
+
+ List<Future<File>> futures =
fetcher.snapshotPrefetchFuturesForTest();
+ assertThat(futures).hasSizeGreaterThanOrEqualTo(2);
+
+ fetcher.close();
+
+ for (Future<File> f : futures) {
+ assertThat(f.isDone()).isTrue();
+ boolean exceptional = false;
+ try {
+ f.get();
+ } catch (CancellationException | ExecutionException expected) {
+ exceptional = true;
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ }
+ assertThat(f.isCancelled() || exceptional).isTrue();
+ }
+ } finally {
+ remoteLogStorage.fetchLogDataBarrier = null;
+ release.countDown();
+ fetcher.close();
+ }
+ }
+
+ //
======================================================================================
+ // Integration tests: slow / failing remote storage via
TestingRemoteLogStorage hooks.
+ //
======================================================================================
+
+ @Test
+ void testSlowDownloadOverlappedWithConsumption() throws Exception {
+ TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
+ makeLogTableAsLeader(tb, false);
+ Replica replica = replicaManager.getReplicaOrException(tb);
+ LogTablet logTablet = replica.getLogTablet();
+ addMultiSegmentsToLogTablet(logTablet, 8);
+
+ remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+
+ List<RemoteLogSegment> segments =
remoteLogManager.relevantRemoteLogSegments(tb, 0L);
+ assertThat(segments).hasSizeGreaterThanOrEqualTo(4);
+ long remoteEndOffset = segments.get(segments.size() -
1).remoteLogEndOffset();
+
+ remoteLogStorage.fetchLogDataDelayMs.set(200L);
+
+ File logTabletDir = logTablet.getLogDir();
+ // Three concurrent slow downloads should overlap with consumption.
+ try (RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir, 3, 3)) {
+ Path tempDir = fetcher.getTempDir();
+ long startNanos = System.nanoTime();
+
+ Iterator<LogRecordBatch> iterator = fetcher.fetch(0,
remoteEndOffset).iterator();
+ assertThat(iterator.hasNext()).isTrue();
+ iterator.next();
+
+ long firstBatchElapsedMs =
+ java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(
+ System.nanoTime() - startNanos);
+
+ // Window should hold >= 3 .log files, proving parallel downloads.
+ waitUntilLogFileCountAtLeast(tempDir, 3);
+ assertThat(countLogFiles(tempDir)).isGreaterThanOrEqualTo(3);
+
+ int batchCount = 1;
+ while (iterator.hasNext()) {
+ iterator.next();
+ batchCount++;
+ }
+ assertThat(batchCount).isGreaterThan(10);
+
+ // Serial downloads would need 4 * 200ms; parallel keeps
first-batch wait ~200ms.
+ assertThat(firstBatchElapsedMs).isLessThan(2_000L);
+ } finally {
+ remoteLogStorage.fetchLogDataDelayMs.set(0L);
+ }
+ }
+
+ @Test
+ void testPersistentDownloadFailureTriggersRetryAndPropagates() throws
Exception {
+ TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
+ makeLogTableAsLeader(tb, false);
+ Replica replica = replicaManager.getReplicaOrException(tb);
+ LogTablet logTablet = replica.getLogTablet();
+ addMultiSegmentsToLogTablet(logTablet, 5);
+
+ remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+
+ List<RemoteLogSegment> segments =
remoteLogManager.relevantRemoteLogSegments(tb, 0L);
+ assertThat(segments).isNotEmpty();
+ long remoteEndOffset = segments.get(segments.size() -
1).remoteLogEndOffset();
+
+ // Force every fetchLogData to fail once uploads are done.
+ remoteLogStorage.fetchLogDataAlwaysFail.set(true);
+
+ File logTabletDir = logTablet.getLogDir();
+ try (RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir, 1, 1)) {
+ int countBefore = remoteLogStorage.fetchLogDataInvocationCount();
+
+ // Head segment exhausts retries then fails.
+ assertThatThrownBy(() -> fetcher.fetch(0,
remoteEndOffset).iterator().hasNext())
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Failed to fetch remote log
segment");
+
+ int countAfter = remoteLogStorage.fetchLogDataInvocationCount();
+ // Eager prefetch may perform one extra invocation.
+ assertThat(countAfter - countBefore)
+
.isGreaterThanOrEqualTo(RemoteLogFetcher.DOWNLOAD_MAX_RETRIES + 1);
+ } finally {
+ remoteLogStorage.fetchLogDataAlwaysFail.set(false);
+ }
+ }
+
+ @Test
+ void testTransientDownloadFailureRecoversOnRetry() throws Exception {
+ TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
+ makeLogTableAsLeader(tb, false);
+ Replica replica = replicaManager.getReplicaOrException(tb);
+ LogTablet logTablet = replica.getLogTablet();
+ addMultiSegmentsToLogTablet(logTablet, 5);
+
+ remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+
+ List<RemoteLogSegment> segments =
remoteLogManager.relevantRemoteLogSegments(tb, 0L);
+ assertThat(segments).isNotEmpty();
+ long remoteEndOffset = segments.get(segments.size() -
1).remoteLogEndOffset();
+
+ // First 2 calls fail, 3rd succeeds — retry loop must recover.
+ remoteLogStorage.fetchLogDataFailFirstN.set(2);
+
+ File logTabletDir = logTablet.getLogDir();
+ try (RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir, 1, 1)) {
+ int batchCount = 0;
+ long prevNextOffset = 0;
for (LogRecordBatch batch : fetcher.fetch(0, remoteEndOffset)) {
- // Just iterate through.
+
assertThat(batch.baseLogOffset()).isGreaterThanOrEqualTo(prevNextOffset);
+ prevNextOffset = batch.nextLogOffset();
+ batchCount++;
}
- // Temp dir should contain downloaded log files.
- File[] downloadedFiles = tempDir.toFile().listFiles();
- assertThat(downloadedFiles).isNotNull();
- assertThat(downloadedFiles.length).isGreaterThan(0);
- // All files should end with .log.
- for (File file : downloadedFiles) {
- assertThat(file.getName()).endsWith(".log");
+ assertThat(batchCount).isGreaterThan(0);
+
assertThat(remoteLogStorage.fetchLogDataInvocationCount()).isGreaterThanOrEqualTo(3);
+ }
+ }
+
+ @Test
+ void testCloseInterruptsOngoingDownload() throws Exception {
+ TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
+ makeLogTableAsLeader(tb, false);
+ Replica replica = replicaManager.getReplicaOrException(tb);
+ LogTablet logTablet = replica.getLogTablet();
+ addMultiSegmentsToLogTablet(logTablet, 5);
+
+ remoteLogTaskScheduler.triggerPeriodicScheduledTasks();
+
+ List<RemoteLogSegment> segments =
remoteLogManager.relevantRemoteLogSegments(tb, 0L);
+ assertThat(segments).isNotEmpty();
+ long remoteEndOffset = segments.get(segments.size() -
1).remoteLogEndOffset();
+
+ // Block workers in fetchLogData; the only way out is
Thread.interrupt().
+ int expectedInFlight = 2;
+ CountDownLatch entered = new CountDownLatch(expectedInFlight);
+ CountDownLatch release = new CountDownLatch(1);
+ CountDownLatch finished = new CountDownLatch(expectedInFlight);
+
+ remoteLogStorage.fetchLogDataBarrier =
+ () -> {
+ entered.countDown();
+ try {
+ release.await();
+ } finally {
+ finished.countDown();
+ }
+ };
+
+ File logTabletDir = logTablet.getLogDir();
+ RemoteLogFetcher fetcher = newFetcher(tb, logTabletDir, 2, 2);
+ try {
+ fetcher.fetch(0, remoteEndOffset);
+
+ assertThat(entered.await(30, TimeUnit.SECONDS))
+ .as("prefetch workers should have entered fetchLogData")
+ .isTrue();
+
+ List<Future<File>> inflight =
fetcher.snapshotPrefetchFuturesForTest();
+ assertThat(inflight).isNotEmpty();
+
+ long closeStart = System.nanoTime();
+ fetcher.close();
+ long closeMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
closeStart);
+
+ // release is never counted down — finished reaching zero proves
interrupt.
+ assertThat(finished.await(10, TimeUnit.SECONDS))
+ .as("close() should have interrupted the blocked prefetch
workers")
+ .isTrue();
+
+ assertThat(closeMs).isLessThan(5_000L);
+
+ for (Future<File> f : inflight) {
+ assertThat(f.isDone()).isTrue();
+ boolean exceptional = false;
+ try {
+ f.get();
+ } catch (CancellationException | ExecutionException expected) {
+ exceptional = true;
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ }
+ assertThat(f.isCancelled() || exceptional).isTrue();
}
+ } finally {
+ remoteLogStorage.fetchLogDataBarrier = null;
+ release.countDown();
+ fetcher.close();
+ }
+ }
+
+ private static void waitUntilLogFileCountAtLeast(Path dir, int
expectedCount) throws Exception {
+ long deadline = System.currentTimeMillis() + 5_000L;
+ while (System.currentTimeMillis() < deadline) {
+ if (Files.exists(dir) && countLogFiles(dir) >= expectedCount) {
+ return;
+ }
+ Thread.sleep(50L);
+ }
+
+ assertThat(countLogFiles(dir)).isGreaterThanOrEqualTo(expectedCount);
+ }
+
+ private static long countLogFiles(Path dir) throws IOException {
+ if (!Files.exists(dir)) {
+ return 0L;
+ }
+
+ try (java.util.stream.Stream<Path> paths = Files.list(dir)) {
+ return paths.filter(path ->
path.getFileName().toString().endsWith(".log")).count();
}
}
}
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/TestingRemoteLogStorage.java
b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/TestingRemoteLogStorage.java
index ef0b63336..a946e9dd0 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/TestingRemoteLogStorage.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/TestingRemoteLogStorage.java
@@ -23,9 +23,14 @@ import org.apache.fluss.fs.FsPath;
import org.apache.fluss.remote.RemoteLogSegment;
import java.io.IOException;
+import java.io.InputStream;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
/**
* A testing implementation of {@link
org.apache.fluss.server.log.remote.RemoteLogStorage} which can
@@ -44,6 +49,54 @@ public class TestingRemoteLogStorage extends
DefaultRemoteLogStorage {
private final AtomicInteger copySegmentCount = new AtomicInteger(0);
+ /**
+ * When set to a positive value, each call to {@link #fetchLogData} will
sleep for the given
+ * number of milliseconds before delegating to the real implementation.
Used by integration
+ * tests that need to simulate a slow remote storage backend.
+ */
+ public final AtomicLong fetchLogDataDelayMs = new AtomicLong(0L);
+
+ /**
+ * When set to a positive value N, the first N {@link #fetchLogData} calls
(across <b>all</b>
+ * segments) will throw a {@link RemoteStorageException}. Subsequent calls
behave normally.
+ * Useful to exercise the exponential-backoff retry loop inside {@code
+ * RemoteLogFetcher#downloadSegmentWithRetry}.
+ */
+ public final AtomicInteger fetchLogDataFailFirstN = new AtomicInteger(0);
+
+ /**
+ * When set to {@code true}, every {@link #fetchLogData} call throws a
{@link
+ * RemoteStorageException}. Used to validate that retries are exhausted
and the failure is
+ * correctly propagated.
+ */
+ public final AtomicBoolean fetchLogDataAlwaysFail = new
AtomicBoolean(false);
+
+ /**
+ * Per-segment failure budget. For each entry {@code (segmentId, N)}, the
next N {@link
+ * #fetchLogData} calls that target that segment throw a {@link
RemoteStorageException};
+ * subsequent calls for the same segment behave normally. Useful for tests
that need a specific
+ * segment's prefetch to exhaust its retries (i.e. {@code
DOWNLOAD_MAX_RETRIES + 1} attempts) so
+ * the prefetch path raises {@code ExecutionException} which is then
propagated directly by
+ * {@code RemoteLogFetcher#fetchSegmentFile} — without poisoning unrelated
segments.
+ */
+ public final Map<UUID, AtomicInteger> fetchLogDataFailureBudget = new
ConcurrentHashMap<>();
+
+ private final AtomicInteger fetchLogDataCount = new AtomicInteger(0);
+
+ /**
+ * Optional test hook invoked inside {@link #fetchLogData} on the worker
thread. Tests can block
+ * the worker on their own latches and assert real interruption. If it
throws {@link
+ * InterruptedException}, the worker's interrupt flag is restored and a
{@link
+ * RemoteStorageException} is propagated. Default null preserves existing
behavior.
+ */
+ public volatile FetchLogDataBarrier fetchLogDataBarrier = null;
+
+ /** Interruptible test hook for {@link
TestingRemoteLogStorage#fetchLogDataBarrier}. */
+ @FunctionalInterface
+ public interface FetchLogDataBarrier {
+ void run() throws InterruptedException;
+ }
+
public TestingRemoteLogStorage(Configuration conf, ExecutorService
ioExecutor)
throws IOException {
super(conf, ioExecutor);
@@ -70,4 +123,56 @@ public class TestingRemoteLogStorage extends
DefaultRemoteLogStorage {
}
return super.writeRemoteLogManifestSnapshot(manifest);
}
+
+ @Override
+ public InputStream fetchLogData(RemoteLogSegment remoteLogSegment)
+ throws RemoteStorageException {
+ int idx = fetchLogDataCount.getAndIncrement();
+ if (fetchLogDataAlwaysFail.get()) {
+ throw new RemoteStorageException(
+ "Simulated persistent fetchLogData failure for segment "
+ + remoteLogSegment.remoteLogSegmentId());
+ }
+ int failFirstN = fetchLogDataFailFirstN.get();
+ if (idx < failFirstN) {
+ throw new RemoteStorageException(
+ "Simulated transient fetchLogData failure, attempt "
+ + (idx + 1)
+ + "/"
+ + failFirstN);
+ }
+ AtomicInteger budget =
fetchLogDataFailureBudget.get(remoteLogSegment.remoteLogSegmentId());
+ if (budget != null && budget.getAndUpdate(n -> n > 0 ? n - 1 : 0) > 0)
{
+ throw new RemoteStorageException(
+ "Simulated per-segment fetchLogData failure for segment "
+ + remoteLogSegment.remoteLogSegmentId()
+ + " (budget remaining: "
+ + budget.get()
+ + ")");
+ }
+ FetchLogDataBarrier barrier = fetchLogDataBarrier;
+ if (barrier != null) {
+ try {
+ barrier.run();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RemoteStorageException("Interrupted at fetchLogData
test barrier", e);
+ }
+ }
+ long delayMs = fetchLogDataDelayMs.get();
+ if (delayMs > 0L) {
+ try {
+ Thread.sleep(delayMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RemoteStorageException("Interrupted while simulating
slow download", e);
+ }
+ }
+ return super.fetchLogData(remoteLogSegment);
+ }
+
+ /** Returns how many times {@link #fetchLogData} has been invoked since
construction. */
+ public int fetchLogDataInvocationCount() {
+ return fetchLogDataCount.get();
+ }
}
diff --git a/website/docs/maintenance/configuration.md
b/website/docs/maintenance/configuration.md
index d681ae0bd..ea3fec7fd 100644
--- a/website/docs/maintenance/configuration.md
+++ b/website/docs/maintenance/configuration.md
@@ -173,6 +173,8 @@ during the Fluss cluster working.
| kv.rocksdb.bloom-filter.block-based-mode | Boolean | false
| If true, RocksDB will use block-based filter instead of
full filter, this only take effect when bloom filter is used. The default value
is `false`.
[...]
| kv.rocksdb.shared-rate-limiter-bytes-per-sec | MemorySize |
Long.MAX_VALUE | The bytes per second rate limit for RocksDB
flush and compaction operations shared across all RocksDB instances on the
TabletServer. The rate limiter is always enabled. The default value is
Long.MAX_VALUE (effectively unlimited). Set to a lower value (e.g., 100MB) to
limit the rate. This configuration can be updated dynamically without server
restart. See [Updating Configs](operations/updat [...]
| kv.recover.log-record-batch.max-size | MemorySize | 16mb
| The max fetch size for fetching log to apply to kv during
recovering kv.
[...]
+| kv.recover.remote-log.prefetch-num | Integer | 4
| The maximum number of remote log segments that can be
downloaded but not yet consumed during KV recovery. A larger value overlaps
more remote storage downloads with consumption, at the cost of extra local disk
usage. Setting to 1 keeps the historical behavior (one-step prefetch). The
default value is 4.
[...]
+| kv.recover.remote-log.download-threads | Integer | 3
| The number of threads used to download remote log segments
during KV recovery. Should be less than or equal to
`kv.recover.remote-log.prefetch-num`. The default value is 3.
[...]
| kv.scanner.ttl | Duration | 10min
| The time-to-live for an idle KV scanner session on the
server. A scanner that has not received a request within this duration will be
automatically expired and its resources released. The default value is 10
minutes.
[...]
| kv.scanner.expiration-interval | Duration | 30s
| The interval at which the server checks for and removes
expired KV scanner sessions. The default value is 30 seconds.
[...]
| kv.scanner.max-per-bucket | Integer | 8
| The maximum number of concurrent KV scanner sessions
allowed per bucket. New scan requests that exceed this limit will be rejected
with an error. The default value is 8.
[...]
