This is an automated email from the ASF dual-hosted git repository.
loserwang1024 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 e176dbb69 [client] cancel pending log download when closing remote log
downloader. (#2786)
e176dbb69 is described below
commit e176dbb69f1094ad97c18ee2c2574c454c4686a2
Author: Hongshun Wang <[email protected]>
AuthorDate: Tue May 26 15:32:23 2026 +0800
[client] cancel pending log download when closing remote log downloader.
(#2786)
---
.../table/scanner/log/RemoteLogDownloader.java | 103 ++++++++++++++-------
.../table/scanner/log/RemoteLogDownloaderTest.java | 91 ++++++++++++++++++
2 files changed, 160 insertions(+), 34 deletions(-)
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloader.java
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloader.java
index b30458983..15bc1f631 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloader.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloader.java
@@ -85,6 +85,8 @@ public class RemoteLogDownloader implements Closeable {
private final long pollTimeout;
+ private volatile boolean closed = false;
+
public RemoteLogDownloader(
TablePath tablePath,
Configuration conf,
@@ -145,6 +147,11 @@ public class RemoteLogDownloader implements Closeable {
// blocks until there is capacity (the fetched file is consumed)
prefetchSemaphore.acquire();
+ if (closed) {
+ prefetchSemaphore.release();
+ return;
+ }
+
// wait until there is a remote fetch request
RemoteLogDownloadRequest request = segmentsToFetch.poll(pollTimeout,
TimeUnit.MILLISECONDS);
if (request == null) {
@@ -163,41 +170,57 @@ public class RemoteLogDownloader implements Closeable {
long startTime = System.currentTimeMillis();
// download the remote file to local
- remoteFileDownloader
- .downloadFileAsync(fsPathAndFileName, localLogDir)
- .whenComplete(
- (bytes, throwable) -> {
- if (throwable != null) {
- LOG.error(
- "Failed to download remote log
segment file {} for table bucket {}.",
- fsPathAndFileName.getFileName(),
- tableBucket,
-
ExceptionUtils.stripExecutionException(throwable));
- // release the semaphore for the failed
request
- prefetchSemaphore.release();
- // add back the request to the queue,
- // so we do not complete the
request.future here
- segmentsToFetch.add(request);
-
scannerMetricGroup.remoteFetchErrorCount().inc();
- } else {
- LOG.info(
- "Successfully downloaded remote
log segment file {} to local for "
- + "table bucket {} cost {}
ms.",
- fsPathAndFileName.getFileName(),
- tableBucket,
- System.currentTimeMillis() -
startTime);
- File localFile =
- new File(
- localLogDir.toFile(),
-
fsPathAndFileName.getFileName());
-
scannerMetricGroup.remoteFetchBytes().inc(bytes);
- request.future.complete(localFile);
- }
- });
+ CompletableFuture<Long> completableFuture =
+ remoteFileDownloader.downloadFileAsync(fsPathAndFileName,
localLogDir);
+ completableFuture.whenComplete(
+ (bytes, throwable) -> {
+ if (closed) {
+ LOG.warn(
+ "RemoteLogDownloader closed when remote
log segment file {} for table bucket {}.",
+ fsPathAndFileName.getFileName(),
+ tableBucket);
+ // In-flight download completed after close.
Cancel the
+ // external future so consumers blocked on .get()
are
+ // unblocked, release the semaphore (no consumer
will
+ // recycle) and clean up any orphan files/dirs the
+ // download may have recreated via
Files.createDirectories
+ // — even on failure, since the directory may
already have
+ // been re-created before the error occurred.
+ request.future.cancel(false);
+ prefetchSemaphore.release();
+ deleteDirectoryQuietly(localLogDir.toFile());
+ return;
+ }
+ if (throwable != null) {
+ LOG.error(
+ "Failed to download remote log segment
file {} for table bucket {}.",
+ fsPathAndFileName.getFileName(),
+ tableBucket,
+
ExceptionUtils.stripExecutionException(throwable));
+ prefetchSemaphore.release();
+ segmentsToFetch.add(request);
+ scannerMetricGroup.remoteFetchErrorCount().inc();
+ } else {
+ LOG.info(
+ "Successfully downloaded remote log
segment file {} to local for "
+ + "table bucket {} cost {} ms.",
+ fsPathAndFileName.getFileName(),
+ tableBucket,
+ System.currentTimeMillis() - startTime);
+ File localFile =
+ new File(localLogDir.toFile(),
fsPathAndFileName.getFileName());
+ scannerMetricGroup.remoteFetchBytes().inc(bytes);
+ request.future.complete(localFile);
+ }
+ });
} catch (Throwable t) {
prefetchSemaphore.release();
- // add back the request to the queue
- segmentsToFetch.add(request);
+ // only re-queue the request if the downloader is still active
+ if (!closed) {
+ segmentsToFetch.add(request);
+ } else {
+ request.future.cancel(false);
+ }
scannerMetricGroup.remoteFetchErrorCount().inc();
// log the error and continue instead of shutdown the download
thread
LOG.error("Failed to download remote log segment for table bucket
{}.", tableBucket, t);
@@ -226,15 +249,27 @@ public class RemoteLogDownloader implements Closeable {
@Override
public void close() throws IOException {
+ closed = true;
+ // Drain pending requests and cancel their futures to unblock waiting
consumers.
+ drainAndCancelRequests();
try {
downloadThread.shutdown();
} catch (InterruptedException e) {
// ignore
}
-
+ // In-flight downloads on the shared pool may still finish and recreate
+ // localLogDir; the whenComplete callback will re-delete when it sees
closed.
deleteDirectoryQuietly(localLogDir.toFile());
}
+ /** Drains all pending requests and cancels their futures. */
+ private void drainAndCancelRequests() {
+ RemoteLogDownloadRequest request;
+ while ((request = segmentsToFetch.poll()) != null) {
+ request.future.cancel(false);
+ }
+ }
+
@VisibleForTesting
Semaphore getPrefetchSemaphore() {
return prefetchSemaphore;
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloaderTest.java
b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloaderTest.java
index 407756a7e..8984f8fc6 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloaderTest.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloaderTest.java
@@ -48,6 +48,9 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import static org.apache.fluss.record.TestData.DATA1_PHYSICAL_TABLE_PATH;
@@ -282,6 +285,94 @@ class RemoteLogDownloaderTest {
assertThat(results).isEqualTo(expected);
}
+ /**
+ * Tests that close() properly cleans up the local directory even when
there are simultaneously:
+ * (1) already-downloaded files on disk, (2) in-flight downloads still
running on the shared
+ * pool, and (3) pending requests in the queue. After close() the local
directory must not exist
+ * — including after in-flight downloads finish and potentially recreate
it via
+ * Files.createDirectories.
+ */
+ @Test
+ void testCloseWithInFlightAndPendingDownloads() throws Exception {
+ // Latch to block in-flight downloads until explicitly released after
close().
+ CountDownLatch blockLatch = new CountDownLatch(1);
+ // Latch to know when 2 in-flight downloads have entered the blocked
state.
+ CountDownLatch inFlightStarted = new CountDownLatch(2);
+ // Latch to know when 2 in-flight downloads are finished.
+ CountDownLatch inFlightFinished = new CountDownLatch(2);
+
+ class BlockingFileDownloader extends RemoteFileDownloader {
+ private final AtomicInteger enteredCount = new AtomicInteger(0);
+
+ BlockingFileDownloader(int threadNum) {
+ super(threadNum);
+ }
+
+ @Override
+ protected long downloadFile(Path targetFilePath, FsPath
remoteFilePath)
+ throws IOException {
+ int count = enteredCount.incrementAndGet();
+ if (count > 1) {
+ // Block the 2nd and 3rd downloads to simulate in-flight
state.
+ inFlightStarted.countDown();
+ try {
+ blockLatch.await();
+ } catch (InterruptedException e) {
+ throw new IOException("Interrupted while blocking", e);
+ }
+ }
+ long downloadBytes = super.downloadFile(targetFilePath,
remoteFilePath);
+ inFlightFinished.countDown();
+ return downloadBytes;
+ }
+ }
+
+ // prefetch=3: downloads 0,1,2 start; downloads 3,4 remain in the
queue.
+ conf.set(ConfigOptions.CLIENT_SCANNER_REMOTE_LOG_PREFETCH_NUM, 3);
+ BlockingFileDownloader fileDownloader = new BlockingFileDownloader(4);
+ RemoteLogDownloader downloader =
+ new RemoteLogDownloader(
+ DATA1_TABLE_PATH, conf, fileDownloader,
scannerMetricGroup, 10L);
+
+ try {
+ TableBucket tb = new TableBucket(DATA1_TABLE_ID, 0);
+ List<RemoteLogSegment> segments =
+ buildRemoteLogSegmentList(tb, DATA1_PHYSICAL_TABLE_PATH,
5, conf, 10);
+ FsPath tabletDir = remoteLogTabletDir(remoteLogDir,
DATA1_PHYSICAL_TABLE_PATH, tb);
+ List<RemoteLogDownloadFuture> futures =
+ requestRemoteLogs(downloader, tabletDir, segments);
+
+ downloader.start();
+
+ // Wait for the first download to complete (already-downloaded
file on disk) and 2
+ // in-flight downloads to enter the blocked state.
+ // At this point: 1 file downloaded, 2 blocked in-flight, 2
pending.
+ retry(Duration.ofMinutes(1), () ->
assertThat(futures.get(0).isDone()).isTrue());
+ assertThat(inFlightStarted.await(30, TimeUnit.SECONDS)).isTrue();
+ Path localLogDir = downloader.getLocalLogDir();
+ assertThat(localLogDir.toFile().exists()).isTrue();
+
+ // Close the downloader.
+ downloader.close();
+
+ // Pending futures (segments 3, 4) should be cancelled immediately.
+ assertThat(futures.get(3).isDone()).isTrue();
+ assertThat(futures.get(4).isDone()).isTrue();
+
+ // Wait for 2 in-flight downloads finished.
+ blockLatch.countDown();
+ assertThat(inFlightFinished.await(30, TimeUnit.SECONDS)).isTrue();
+ retry(Duration.ofMinutes(1), () ->
assertThat(futures.get(1).isDone()).isTrue());
+
+ // Verify that ultimately the local directory does not exist.
+ retry(Duration.ofMinutes(1), () ->
assertThat(localLogDir.toFile().exists()).isFalse());
+ } finally {
+ blockLatch.countDown(); // ensure latch is released even on test
failure
+ IOUtils.closeQuietly(downloader);
+ IOUtils.closeQuietly(fileDownloader);
+ }
+ }
+
private RemoteLogDownloadRequest createDownloadRequest(
TableBucket tableBucket, long startOffset, long maxTimestamp) {
RemoteLogSegment remoteLogSegment =
