From 465d2066cc21d253ef76fadc6e4c6cf71a0dda63 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=96=B9=E6=99=93=E5=85=B5?= Date: Fri, 31 Jul 2026 13:45:10 +0800 Subject: [PATCH 1/4] [client] Prevent EOF during concurrent remote log downloads - Recheck fetchable buckets under the fetcher lock to avoid stale duplicate requests. - Download into unique temporary files and atomically publish complete segments. - Add deterministic regressions for stale fetch snapshots and open-reader safety. --- .../table/scanner/RemoteFileDownloader.java | 34 ++++--- .../client/table/scanner/log/LogFetcher.java | 7 +- .../table/scanner/log/LogFetcherTest.java | 83 ++++++++++++++++ .../scanner/log/RemoteLogDownloaderTest.java | 98 +++++++++++++++++++ 4 files changed, 207 insertions(+), 15 deletions(-) diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/RemoteFileDownloader.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/RemoteFileDownloader.java index 01269f0c4c..efd1ba1e6b 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/RemoteFileDownloader.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/RemoteFileDownloader.java @@ -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; @@ -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; @@ -88,21 +88,31 @@ public CompletableFuture downloadFileAsync( * downloaded bytes. */ protected long downloadFile(Path targetFilePath, FsPath remoteFilePath) throws IOException { - List 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); + IOException failure = new IOException(ex); + if (temporaryFile != null) { + try { + Files.deleteIfExists(temporaryFile); + } catch (IOException cleanupException) { + failure.addSuppressed(cleanupException); + } + } + throw failure; } } diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java index a178ab2061..1507de6ad9 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java @@ -212,9 +212,10 @@ public void sendFetches() { List fetchable = fetchableBuckets(); checkAndUpdateMetadata(fetchable); 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 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 fetchRequestMap = + prepareFetchLogRequests(fetchableBuckets()); fetchRequestMap.forEach( (nodeId, fetchLogRequest) -> { LOG.debug("Adding pending request for node id {}", nodeId); diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherTest.java b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherTest.java index 564cbb06ea..5c99d5fc88 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherTest.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherTest.java @@ -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; @@ -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; @@ -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 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 defaultRequestMap = @@ -227,19 +259,70 @@ public CompletableFuture fetchLog(FetchLogRequest request) { private static class DelayedTabletServerGateway extends TestingTabletServerGateway { private final CompletableFuture responseFuture = new CompletableFuture<>(); + private final AtomicInteger requestCount = new AtomicInteger(); private FetchLogResponse response; @Override public CompletableFuture 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()); } 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 77f6dc998e..b65253e4c4 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 @@ -23,9 +23,14 @@ 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.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; @@ -148,6 +153,99 @@ void testPrefetchNum() throws Exception { } } + @Test + void testDuplicateRemoteLogDownloadDoesNotBreakOpenReader() throws Exception { + CountDownLatch duplicateDownloadCopyStarted = new CountDownLatch(1); + CountDownLatch continueDuplicateDownload = new CountDownLatch(1); + + class BlockingRemoteFileDownloader extends RemoteFileDownloader { + private final AtomicInteger downloadCount = new AtomicInteger(); + + private BlockingRemoteFileDownloader() { + super(1); + } + + @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); + } + } + + BlockingRemoteFileDownloader fileDownloader = new BlockingRemoteFileDownloader(); + 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 testDiscardQueuedDownload() throws Exception { RemoteFileDownloader fileDownloader = new RemoteFileDownloader(1); From 9b2874f3101e39b939219bf9ff8412cfac995c90 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=96=B9=E6=99=93=E5=85=B5?= Date: Tue, 4 Aug 2026 15:21:25 +0800 Subject: [PATCH 2/4] [client] Simplify fetchable bucket metadata update - Inline the one-use bucket list before metadata refresh. - Keep the locked post-refresh recomputation unchanged. --- .../org/apache/fluss/client/table/scanner/log/LogFetcher.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java index 1507de6ad9..77a8504ca3 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java @@ -209,8 +209,7 @@ public ArrowScanRecords collectArrowFetch() { * have an in-flight fetch or pending fetch data. */ public void sendFetches() { - List fetchable = fetchableBuckets(); - checkAndUpdateMetadata(fetchable); + checkAndUpdateMetadata(fetchableBuckets()); synchronized (this) { // Recompute after metadata update because response callbacks can populate the buffer. // Don't perform heavy I/O operations or synchronous waits here. From 1953097707312495fdc828fd19409a60cd954b32 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=96=B9=E6=99=93=E5=85=B5?= Date: Tue, 4 Aug 2026 16:44:38 +0800 Subject: [PATCH 3/4] [client] Clean up failed remote download temp files - Move temporary-file deletion to finally so errors cannot bypass cleanup. - Add an out-of-memory regression through the asynchronous downloader path. --- .../table/scanner/RemoteFileDownloader.java | 11 ++--- .../scanner/log/RemoteLogDownloaderTest.java | 40 +++++++++++++++++++ 2 files changed, 44 insertions(+), 7 deletions(-) diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/RemoteFileDownloader.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/RemoteFileDownloader.java index efd1ba1e6b..8649947c9b 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/RemoteFileDownloader.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/RemoteFileDownloader.java @@ -104,15 +104,12 @@ protected long downloadFile(Path targetFilePath, FsPath remoteFilePath) throws I FileUtils.atomicMoveWithFallback(temporaryFile, targetFilePath, false); return downloadBytes; } catch (Exception ex) { - IOException failure = new IOException(ex); + throw new IOException(ex); + } finally { if (temporaryFile != null) { - try { - Files.deleteIfExists(temporaryFile); - } catch (IOException cleanupException) { - failure.addSuppressed(cleanupException); - } + Path fileToDelete = temporaryFile; + IOUtils.closeQuietly(() -> Files.deleteIfExists(fileToDelete)); } - throw failure; } } 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 b65253e4c4..dd59447dee 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 @@ -27,6 +27,7 @@ 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; @@ -41,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; @@ -67,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 { @@ -246,6 +249,43 @@ public FileSystem getFileSystem() { } } + @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); From 6a44c98919aec79cba2ae1de1ed0f57cb67b256d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=96=B9=E6=99=93=E5=85=B5?= Date: Tue, 4 Aug 2026 19:11:31 +0800 Subject: [PATCH 4/4] [client] Improve remote download cleanup diagnostics - Log a warning when temporary download file cleanup fails. - Move the duplicate-download test helper out of the test body. --- .../table/scanner/RemoteFileDownloader.java | 4 +- .../scanner/log/RemoteLogDownloaderTest.java | 105 ++++++++++-------- 2 files changed, 59 insertions(+), 50 deletions(-) diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/RemoteFileDownloader.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/RemoteFileDownloader.java index 8649947c9b..8e029ae159 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/RemoteFileDownloader.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/RemoteFileDownloader.java @@ -108,7 +108,9 @@ protected long downloadFile(Path targetFilePath, FsPath remoteFilePath) throws I } finally { if (temporaryFile != null) { Path fileToDelete = temporaryFile; - IOUtils.closeQuietly(() -> Files.deleteIfExists(fileToDelete)); + IOUtils.closeQuietly( + () -> Files.deleteIfExists(fileToDelete), + "temporary download file " + fileToDelete); } } } 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 dd59447dee..8e2fd892e8 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 @@ -161,55 +161,9 @@ void testDuplicateRemoteLogDownloadDoesNotBreakOpenReader() throws Exception { CountDownLatch duplicateDownloadCopyStarted = new CountDownLatch(1); CountDownLatch continueDuplicateDownload = new CountDownLatch(1); - class BlockingRemoteFileDownloader extends RemoteFileDownloader { - private final AtomicInteger downloadCount = new AtomicInteger(); - - private BlockingRemoteFileDownloader() { - super(1); - } - - @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); - } - } - - BlockingRemoteFileDownloader fileDownloader = new BlockingRemoteFileDownloader(); + BlockingRemoteFileDownloader fileDownloader = + new BlockingRemoteFileDownloader( + duplicateDownloadCopyStarted, continueDuplicateDownload); RemoteLogDownloader downloader = new RemoteLogDownloader( DATA1_TABLE_PATH.toString(), conf, fileDownloader, scannerMetricGroup, 10L); @@ -642,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 =