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..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 @@ -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,30 @@ 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); + if (temporaryFile != null) { + Path fileToDelete = temporaryFile; + IOUtils.closeQuietly( + () -> Files.deleteIfExists(fileToDelete), + "temporary download file " + fileToDelete); + } } } 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..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,12 +209,12 @@ public ArrowScanRecords collectArrowFetch() { * have an in-flight fetch or pending fetch data. */ public void sendFetches() { - List 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 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..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 @@ -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; @@ -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; @@ -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 { @@ -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); @@ -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 =