Skip to content

[client] Prevent EOF during concurrent remote log downloads #3816

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fxbing wants to merge 4 commits into
base: main
Choose a base branch
from
Open

[client] Prevent EOF during concurrent remote log downloads #3816

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
465d206
[client] Prevent EOF during concurrent remote log downloads
Jul 31, 2026
9b2874f
[client] Simplify fetchable bucket metadata update
Aug 4, 2026
1953097
[client] Clean up failed remote download temp files
Aug 4, 2026
6a44c98
[client] Improve remote download cleanup diagnostics
Aug 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 19 additions & 10 deletions fluss-client/src/main/java/org/apache/fluss/client/table/scanner/RemoteFileDownloader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import org.apache.fluss.fs.utils.FileDownloadSpec;
import org.apache.fluss.fs.utils.FileDownloadUtils;
import org.apache.fluss.utils.CloseableRegistry;
import org.apache.fluss.utils.FileUtils;
import org.apache.fluss.utils.IOUtils;
import org.apache.fluss.utils.concurrent.ExecutorThreadFactory;

Expand All @@ -32,7 +33,6 @@
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.List;
import java.util.concurrent.CompletableFuture;
Expand Down Expand Up @@ -88,21 +88,30 @@ public CompletableFuture<Long> downloadFileAsync(
* downloaded bytes.
*/
protected long downloadFile(Path targetFilePath, FsPath remoteFilePath) throws IOException {
List<Closeable> closeableRegistry = new ArrayList<>(2);
Path temporaryFile = null;
try {
FileSystem fileSystem = remoteFilePath.getFileSystem();
FSDataInputStream inputStream = fileSystem.open(remoteFilePath);
closeableRegistry.add(inputStream);

Files.createDirectories(targetFilePath.getParent());
OutputStream outputStream = Files.newOutputStream(targetFilePath);
closeableRegistry.add(outputStream);
temporaryFile =
Files.createTempFile(targetFilePath.getParent(), ".fluss-download-", ".tmp");

FileSystem fileSystem = remoteFilePath.getFileSystem();
long downloadBytes;
try (FSDataInputStream inputStream = fileSystem.open(remoteFilePath);
OutputStream outputStream = Files.newOutputStream(temporaryFile)) {
downloadBytes = IOUtils.copyBytes(inputStream, outputStream, false);
}

return IOUtils.copyBytes(inputStream, outputStream, false);
FileUtils.atomicMoveWithFallback(temporaryFile, targetFilePath, false);
return downloadBytes;
} catch (Exception ex) {
throw new IOException(ex);
} finally {
closeableRegistry.forEach(IOUtils::closeQuietly);
if (temporaryFile != null) {

@loserwang1024 loserwang1024 Aug 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The temporaryFile won't be clear if throw OutOfMemoryError. Thus here need to modified as throw or, close in final scope(same as before)

Path fileToDelete = temporaryFile;
IOUtils.closeQuietly(
() -> Files.deleteIfExists(fileToDelete),
"temporary download file " + fileToDelete);
}
}
}

Expand Down
10 changes: 5 additions & 5 deletions fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,12 +209,12 @@ public ArrowScanRecords collectArrowFetch() {
* have an in-flight fetch or pending fetch data.
*/
public void sendFetches() {
List<TableBucket> fetchable = fetchableBuckets();
checkAndUpdateMetadata(fetchable);
checkAndUpdateMetadata(fetchableBuckets());
synchronized (this) {
// NOTE: Don't perform heavy I/O operations or synchronous waits inside this lock to
// avoid blocking the future complete of FetchLogResponse.
Map<Integer, FetchLogRequest> fetchRequestMap = prepareFetchLogRequests(fetchable);
// Recompute after metadata update because response callbacks can populate the buffer.
// Don't perform heavy I/O operations or synchronous waits here.
Map<Integer, FetchLogRequest> fetchRequestMap =
prepareFetchLogRequests(fetchableBuckets());
fetchRequestMap.forEach(
(nodeId, fetchLogRequest) -> {
LOG.debug("Adding pending request for node id {}", nodeId);
Expand Down
83 changes: 83 additions & 0 deletions fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,12 +23,14 @@
import org.apache.fluss.client.metrics.TestingScannerMetricGroup;
import org.apache.fluss.client.table.scanner.RemoteFileDownloader;
import org.apache.fluss.cluster.BucketLocation;
import org.apache.fluss.cluster.Cluster;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.exception.NotLeaderOrFollowerException;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.SchemaInfo;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.record.LogRecordReadContext;
import org.apache.fluss.rpc.entity.FetchLogResultForBucket;
import org.apache.fluss.rpc.messages.FetchLogRequest;
Expand All @@ -50,6 +52,9 @@
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import static org.apache.fluss.client.metadata.TestingMetadataUpdater.NODE1;
import static org.apache.fluss.client.metadata.TestingMetadataUpdater.NODE2;
Expand Down Expand Up @@ -165,6 +170,33 @@ void testDiscardStaleResponseAfterTableReregistered() {
assertThat(logFetcher.getRegisteredTableCount()).isEqualTo(1);
}

@Test
void testSendFetchesRechecksFetchableBucketsAfterMetadataUpdate() throws Exception {
IOUtils.closeQuietly(logFetcher);
DelayedTabletServerGateway delayedGateway = new DelayedTabletServerGateway();
BlockingMetadataUpdater blockingMetadataUpdater =
new BlockingMetadataUpdater(delayedGateway);
metadataUpdater = blockingMetadataUpdater;
logFetcher = createLogFetcher(new Configuration());

logFetcher.sendFetches();
assertThat(delayedGateway.getRequestCount()).isOne();

blockingMetadataUpdater.invalidateTableMetadata();
CompletableFuture<Void> secondSend = CompletableFuture.runAsync(logFetcher::sendFetches);
try {
assertThat(blockingMetadataUpdater.awaitMetadataUpdateStarted()).isTrue();
delayedGateway.completeResponse();
assertThat(logFetcher.getCompletedFetchesSize()).isOne();
} finally {
blockingMetadataUpdater.continueMetadataUpdate();
delayedGateway.completeResponse();
}

secondSend.get(30, TimeUnit.SECONDS);
assertThat(delayedGateway.getRequestCount()).isOne();
}

@Test
void testPrepareFetchLogRequestWithReadPreference() throws Exception {
Map<Integer, FetchLogRequest> defaultRequestMap =
Expand Down Expand Up @@ -227,19 +259,70 @@ public CompletableFuture<FetchLogResponse> fetchLog(FetchLogRequest request) {
private static class DelayedTabletServerGateway extends TestingTabletServerGateway {
private final CompletableFuture<FetchLogResponse> responseFuture =
new CompletableFuture<>();
private final AtomicInteger requestCount = new AtomicInteger();
private FetchLogResponse response;

@Override
public CompletableFuture<FetchLogResponse> fetchLog(FetchLogRequest request) {
requestCount.incrementAndGet();
response = super.fetchLog(request).join();
return responseFuture;
}

private int getRequestCount() {
return requestCount.get();
}

private void completeResponse() {
responseFuture.complete(response);
}
}

private static class BlockingMetadataUpdater extends TestingMetadataUpdater {
private final CountDownLatch metadataUpdateStarted = new CountDownLatch(1);
private final CountDownLatch continueMetadataUpdate = new CountDownLatch(1);
private final Cluster refreshedCluster;

private BlockingMetadataUpdater(TestTabletServerGateway gateway) {
super(
COORDINATOR,
Arrays.asList(NODE1, NODE2, NODE3),
Collections.singletonMap(DATA1_TABLE_PATH, DATA1_TABLE_INFO),
Collections.singletonMap(1, gateway),
new Configuration());
refreshedCluster = getCluster();
}

@Override
public void updateTableOrPartitionMetadata(TablePath tablePath, Long partitionId) {
metadataUpdateStarted.countDown();
try {
if (!continueMetadataUpdate.await(30, TimeUnit.SECONDS)) {
throw new AssertionError("Timed out waiting to continue metadata update");
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new AssertionError("Interrupted while blocking metadata update", e);
}
updateCluster(refreshedCluster);
}

private void invalidateTableMetadata() {
updateCluster(
getCluster()
.invalidPhysicalTableBucketMeta(
Collections.singleton(PhysicalTablePath.of(DATA1_TABLE_PATH))));
}

private boolean awaitMetadataUpdateStarted() throws InterruptedException {
return metadataUpdateStarted.await(30, TimeUnit.SECONDS);
}

private void continueMetadataUpdate() {
continueMetadataUpdate.countDown();
}
}

private TestingMetadataUpdater initializeMetadataUpdater() {
return initializeMetadataUpdater(new TestingTabletServerGateway());
}
Expand Down
145 changes: 145 additions & 0 deletions ...ient/src/test/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloaderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,9 +23,15 @@
import org.apache.fluss.client.table.scanner.log.RemoteLogDownloader.RemoteLogDownloadRequest;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.fs.FSDataInputStream;
import org.apache.fluss.fs.FSDataInputStreamWrapper;
import org.apache.fluss.fs.FileSystem;
import org.apache.fluss.fs.FsPath;
import org.apache.fluss.fs.FsPathAndFileName;
import org.apache.fluss.fs.local.LocalFileSystem;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.record.FileLogRecords;
import org.apache.fluss.remote.RemoteLogSegment;
import org.apache.fluss.utils.FileUtils;
import org.apache.fluss.utils.IOUtils;
Expand All @@ -36,6 +42,7 @@

import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Duration;
import java.util.ArrayList;
Expand All @@ -62,6 +69,7 @@
import static org.apache.fluss.utils.FlussPaths.remoteLogDir;
import static org.apache.fluss.utils.FlussPaths.remoteLogTabletDir;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

/** Tests for {@link RemoteLogDownloader}. */
class RemoteLogDownloaderTest {
Expand Down Expand Up @@ -148,6 +156,90 @@ void testPrefetchNum() throws Exception {
}
}

@Test
void testDuplicateRemoteLogDownloadDoesNotBreakOpenReader() throws Exception {
CountDownLatch duplicateDownloadCopyStarted = new CountDownLatch(1);
CountDownLatch continueDuplicateDownload = new CountDownLatch(1);

BlockingRemoteFileDownloader fileDownloader =
new BlockingRemoteFileDownloader(
duplicateDownloadCopyStarted, continueDuplicateDownload);
RemoteLogDownloader downloader =
new RemoteLogDownloader(
DATA1_TABLE_PATH.toString(), conf, fileDownloader, scannerMetricGroup, 10L);
try {
TableBucket tableBucket = new TableBucket(DATA1_TABLE_ID, 0);
RemoteLogSegment segment =
buildRemoteLogSegmentList(tableBucket, DATA1_PHYSICAL_TABLE_PATH, 1, conf, 10)
.get(0);
FsPath tabletDir =
remoteLogTabletDir(remoteLogDir, DATA1_PHYSICAL_TABLE_PATH, tableBucket);

RemoteLogDownloadFuture firstDownload = downloader.requestRemoteLog(tabletDir, segment);
downloader.fetchOnce();
retry(Duration.ofSeconds(30), () -> assertThat(firstDownload.isDone()).isTrue());

RemoteLogDownloadFuture duplicateDownload =
downloader.requestRemoteLog(tabletDir, segment);
FileLogRecords openReader = firstDownload.getFileLogRecords(0);
try {
downloader.fetchOnce();
assertThat(duplicateDownloadCopyStarted.await(30, TimeUnit.SECONDS)).isTrue();
assertThat(openReader.batches().iterator().hasNext()).isTrue();
continueDuplicateDownload.countDown();
retry(
Duration.ofSeconds(30),
() -> assertThat(duplicateDownload.isDone()).isTrue());
duplicateDownload.getFileLogRecords(0).closeHandlers();
} finally {
continueDuplicateDownload.countDown();
openReader.closeHandlers();
}

assertThat(FileUtils.listDirectory(downloader.getLocalLogDir())).hasSize(1);
} finally {
IOUtils.closeQuietly(downloader);
IOUtils.closeQuietly(fileDownloader);
}
}

@Test
void testTemporaryFileCleanupOnError() throws Exception {
Path remoteFile = remoteDataDir.toPath().resolve("remote.log");
Files.write(remoteFile, new byte[] {1});
FileSystem failingFileSystem =
new LocalFileSystem() {
@Override
public FSDataInputStream open(FsPath path) throws IOException {
return new FSDataInputStreamWrapper(super.open(path)) {
@Override
public int read(byte[] buffer) {
throw new OutOfMemoryError("test");
}
};
}
};
FsPath remotePath =
new FsPath(remoteFile.toUri()) {
@Override
public FileSystem getFileSystem() {
return failingFileSystem;
}
};

try (RemoteFileDownloader downloader = new RemoteFileDownloader(1)) {
assertThatThrownBy(
() ->
downloader
.downloadFileAsync(
new FsPathAndFileName(remotePath, "local.log"),
localDir.toPath())
.get())
.hasCauseInstanceOf(OutOfMemoryError.class);
}
assertThat(FileUtils.listDirectory(localDir.toPath())).isEmpty();
}

@Test
void testDiscardQueuedDownload() throws Exception {
RemoteFileDownloader fileDownloader = new RemoteFileDownloader(1);
Expand Down Expand Up @@ -504,6 +596,59 @@ protected long downloadFile(Path targetFilePath, FsPath remoteFilePath)
}
}

private static class BlockingRemoteFileDownloader extends RemoteFileDownloader {
private final CountDownLatch duplicateDownloadCopyStarted;
private final CountDownLatch continueDuplicateDownload;
private final AtomicInteger downloadCount = new AtomicInteger();

private BlockingRemoteFileDownloader(
CountDownLatch duplicateDownloadCopyStarted,
CountDownLatch continueDuplicateDownload) {
super(1);
this.duplicateDownloadCopyStarted = duplicateDownloadCopyStarted;
this.continueDuplicateDownload = continueDuplicateDownload;
}

@Override
protected long downloadFile(Path targetFilePath, FsPath remoteFilePath) throws IOException {
if (downloadCount.incrementAndGet() == 2) {
FileSystem blockingFileSystem =
new LocalFileSystem() {
@Override
public FSDataInputStream open(FsPath path) throws IOException {
return new FSDataInputStreamWrapper(super.open(path)) {
@Override
public int read(byte[] buffer) throws IOException {
duplicateDownloadCopyStarted.countDown();
try {
if (!continueDuplicateDownload.await(
30, TimeUnit.SECONDS)) {
throw new IOException(
"Timed out waiting to continue duplicate download");
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IOException(
"Interrupted while blocking duplicate download",
e);
}
return super.read(buffer);
}
};
}
};
remoteFilePath =
new FsPath(remoteFilePath.toUri()) {
@Override
public FileSystem getFileSystem() {
return blockingFileSystem;
}
};
}
return super.downloadFile(targetFilePath, remoteFilePath);
}
}

private RemoteLogDownloadRequest createDownloadRequest(
TableBucket tableBucket, long startOffset, long maxTimestamp) {
RemoteLogSegment remoteLogSegment =
Expand Down
Loading