Re: [PR] [server] Optimize RemoteLogFetcher with async prefetch for recovery [fluss]

["Kaixuan-Duan (via GitHub)" <gi...@apache.org>]

Kaixuan-Duan commented on code in PR #3132:
URL: https://github.com/apache/fluss/pull/3132#discussion_r3176679786


##########
fluss-server/src/main/java/org/apache/fluss/server/kv/RemoteLogFetcher.java:
##########
@@ -333,12 +564,207 @@ private void advance() {
             }
         }
 
+        /**
+         * Obtain the local file for {@code segment}, ideally reusing a prefetched download. On
+         * success the permit of the popped head entry is released and a refill is triggered, which
+         * is the Java analogue of Rust's {@code Drop PrefetchPermit + recycle_notify}.
+         */
+        private File fetchSegmentFile(RemoteLogSegment segment) throws IOException {
+            PrefetchEntry head = prefetchQueue.peekFirst();
+            if (head != null && isSameSegmentId(segment, head.segment)) {
+                prefetchQueue.pollFirst();
+                try {
+                    return head.future.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, fallback to sync download.",
+                            segment.remoteLogSegmentId(),
+                            e.getCause());
+                    return downloadSegmentWithRetry(segment);
+                } finally {
+                    releasePermitOnce(head);
+                    // Window slot released — try to enqueue the next prefetch right away.
+                    fillPrefetchWindow();
+                }
+            }
+
+            // Defensive fallback: this branch is logically unreachable in production.
+            // Both fillPrefetchWindow() and advance() iterate the same pre-filtered `segments`
+            // list (returned by RemoteLogManager#relevantRemoteLogSegments) in strict index
+            // order with identical skip rules, and consumption is single-threaded FIFO. So the
+            // queue head is always the next segment advance() will request.
+            //
+            // We still keep the branch as a safety net for two cases:
+            //  1) tests that inject a mismatched entry via injectPrefetchEntryForTest;
+            //  2) future refactors that accidentally break the FIFO/ordering invariant —
+            //     instead of returning wrong data we drop the stale window and re-download
+            //     the requested segment synchronously, at the cost of wasting the already
+            //     prefetched files.
+            if (head != null) {
+                LOG.debug(
+                        "Prefetch queue head segment {} does not match requested segment {}; "
+                                + "dropping stale window and falling back to sync download.",
+                        head.segment.remoteLogSegmentId(),
+                        segment.remoteLogSegmentId());
+            }
+            drainAndReleasePrefetchQueue();
+            fillPrefetchWindow();
+            return downloadSegmentWithRetry(segment);
+        }
+
+        private boolean isSameSegmentId(RemoteLogSegment left, RemoteLogSegment right) {
+            return left != null
+                    && right != null
+                    && left.remoteLogSegmentId().equals(right.remoteLogSegmentId());
+        }
+
+        /**
+         * Enqueue as many download tasks as the window permits. Non-blocking: if the permit
+         * semaphore is empty we stop and wait for the consumer to release a slot.
+         */
+        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;
+                }
+
+                // Non-blocking permit acquisition. If the window is full, back off and let the
+                // consumer release a permit via releasePermitOnce().
+                if (!prefetchPermits.tryAcquire()) {
+                    return;
+                }
+
+                final RemoteLogSegment target = segment;
+                CompletableFuture<File> future;
+                try {

Review Comment:
   The async worker now calls downloadSegmentWithRetry(target), so transient failures are retried in the worker thread instead of falling back to the recovery thread.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org