diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchBuffer.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchBuffer.java index b6b6e61442..3a2f5d98e6 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchBuffer.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchBuffer.java @@ -18,8 +18,10 @@ package org.apache.fluss.client.table.scanner.log; import org.apache.fluss.annotation.Internal; +import org.apache.fluss.exception.FetchException; import org.apache.fluss.exception.WakeupException; import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.utils.ExceptionUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -73,17 +75,19 @@ public class LogFetchBuffer implements AutoCloseable { @GuardedBy("lock") private @Nullable CompletedFetch nextInLineFetch; + @GuardedBy("lock") + private @Nullable Throwable throwable; + public LogFetchBuffer() { this.completedFetches = new LinkedList<>(); } /** - * Returns {@code true} if there are no completed fetches pending to return to the user. - * - * @return {@code true} if the buffer is empty, {@code false} otherwise + * @return {@code true} if there are no completed fetches pending to return to the user and no + * error has been recorded, {@code false} otherwise */ boolean isEmpty() { - return inLock(lock, completedFetches::isEmpty); + return inLock(lock, () -> completedFetches.isEmpty() && throwable == null); } void pend(PendingFetch pendingFetch) { @@ -100,10 +104,16 @@ void pend(PendingFetch pendingFetch) { * Tries to complete the pending fetches in order, convert them into completed fetches in the * buffer. */ - void tryComplete(TableBucket tableBucket) { + void tryComplete(TableBucket tableBucket, Throwable t) { inLock( lock, () -> { + if (t != null) { + this.throwable = t; + notEmptyCondition.signalAll(); + return; + } + boolean hasCompleted = false; LinkedList pendings = this.pendingFetches.get(tableBucket); while (pendings != null && !pendings.isEmpty()) { @@ -157,12 +167,22 @@ void setNextInLineFetch(@Nullable CompletedFetch nextInLineFetch) { inLock(lock, () -> this.nextInLineFetch = nextInLineFetch); } - CompletedFetch peek() { - return inLock(lock, completedFetches::peek); + CompletedFetch peek() throws FetchException { + return inLock( + lock, + () -> { + checkException(); + return completedFetches.peek(); + }); } - CompletedFetch poll() { - return inLock(lock, completedFetches::poll); + CompletedFetch poll() throws FetchException { + return inLock( + lock, + () -> { + checkException(); + return completedFetches.poll(); + }); } /** @@ -282,6 +302,12 @@ Set pendedBuckets() { return inLock(lock, pendingFetches::keySet); } + void checkException() throws FetchException { + if (throwable != null) { + throw new FetchException(ExceptionUtils.stripCompletionException(throwable)); + } + } + @Override public void close() throws Exception { inLock(lock, () -> retainAll(Collections.emptySet())); 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 2eec56b34c..28cf324eb4 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 @@ -560,7 +560,8 @@ private void pendRemoteFetches( logScannerStatus, isCheckCrcs); logFetchBuffer.pend(pendingFetch); - downloadFuture.onComplete(() -> logFetchBuffer.tryComplete(segment.tableBucket())); + downloadFuture.whenComplete( + (throwable) -> logFetchBuffer.tryComplete(segment.tableBucket(), throwable)); } } diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloadFuture.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloadFuture.java index 57e16989c9..8a68dc409f 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloadFuture.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemoteLogDownloadFuture.java @@ -24,6 +24,7 @@ import java.io.IOException; import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Consumer; /** Represents the future of a remote log download request. */ public class RemoteLogDownloadFuture { @@ -77,4 +78,13 @@ public void discard() { public void onComplete(Runnable callback) { logFileFuture.thenRun(callback); } + + public void whenComplete(Consumer callback) { + logFileFuture.whenComplete( + (file, throwable) -> { + if (!logFileFuture.isCancelled()) { + callback.accept(throwable); + } + }); + } } 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 04c03eb706..cdde01919d 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 @@ -28,6 +28,7 @@ import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.remote.RemoteLogSegment; import org.apache.fluss.utils.ExceptionUtils; +import org.apache.fluss.utils.ExponentialBackoff; import org.apache.fluss.utils.FlussPaths; import org.apache.fluss.utils.concurrent.ShutdownableThread; @@ -62,6 +63,17 @@ public class RemoteLogDownloader implements Closeable { private static final Logger LOG = LoggerFactory.getLogger(RemoteLogDownloader.class); private static final long POLL_TIMEOUT = 5000L; + private static final long RETRY_BACKOFF_INITIAL_MS = 100L; + private static final int RETRY_BACKOFF_MULTIPLIER = 2; + private static final long RETRY_BACKOFF_MAX_MS = 5000L; + private static final double RETRY_BACKOFF_JITTER = 0.25D; + private static final ExponentialBackoff RETRY_BACKOFF = + new ExponentialBackoff( + RETRY_BACKOFF_INITIAL_MS, + RETRY_BACKOFF_MULTIPLIER, + RETRY_BACKOFF_MAX_MS, + RETRY_BACKOFF_JITTER); + private final int maxRetryCount; private final Path localLogDir; @@ -107,6 +119,7 @@ public RemoteLogDownloader( this.remoteFileDownloader = remoteFileDownloader; this.scannerMetricGroup = scannerMetricGroup; this.pollTimeout = pollTimeout; + this.maxRetryCount = conf.getInt(ConfigOptions.CLIENT_SCANNER_REMOTE_LOG_FETCH_MAX_RETRIES); this.prefetchSemaphore = new Semaphore(conf.getInt(ConfigOptions.CLIENT_SCANNER_REMOTE_LOG_PREFETCH_NUM)); // The local tmp dir to store the fetched log segment files, @@ -171,7 +184,21 @@ void fetchOnce() throws Exception { return; } - TableBucket tableBucket = request.getTableBucket(); + downloadRemoteLog(request, maxRetryCount, System.currentTimeMillis()); + } + + private void downloadRemoteLog( + RemoteLogDownloadRequest request, int retryCount, long startTime) { + if (closed || request.future.isCancelled()) { + if (!request.future.isDone()) { + request.future.cancel(false); + } + prefetchSemaphore.release(); + if (closed) { + deleteDirectoryQuietly(localLogDir.toFile()); + } + return; + } try { // 1. cleanup the finished logs first to free up disk space cleanupRemoteLogs(); @@ -180,7 +207,6 @@ void fetchOnce() throws Exception { FsPathAndFileName fsPathAndFileName = request.getFsPathAndFileName(); scannerMetricGroup.remoteFetchRequestCount().inc(); - long startTime = System.currentTimeMillis(); // download the remote file to local CompletableFuture completableFuture = remoteFileDownloader.downloadFileAsync(fsPathAndFileName, localLogDir); @@ -190,7 +216,7 @@ void fetchOnce() throws Exception { LOG.warn( "RemoteLogDownloader closed when remote log segment file {} for table bucket {}.", fsPathAndFileName.getFileName(), - tableBucket); + request.getTableBucket()); // In-flight download completed after close. Cancel the // external future so consumers blocked on .get() are // unblocked, release the semaphore (no consumer will @@ -211,20 +237,13 @@ void fetchOnce() throws Exception { 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(); + handleFetchException(request, throwable, retryCount, startTime); } else { LOG.info( "Successfully downloaded remote log segment file {} to local for " + "table bucket {} cost {} ms.", fsPathAndFileName.getFileName(), - tableBucket, + request.getTableBucket(), System.currentTimeMillis() - startTime); File localFile = new File(localLogDir.toFile(), fsPathAndFileName.getFileName()); @@ -236,16 +255,70 @@ void fetchOnce() throws Exception { } }); } catch (Throwable t) { - prefetchSemaphore.release(); - // only re-queue the request if the downloader is still active - if (!closed && !request.future.isCancelled()) { - segmentsToFetch.add(request); - } else if (!request.future.isDone()) { + if (closed || request.future.isCancelled()) { + if (!request.future.isDone()) { + request.future.cancel(false); + } + prefetchSemaphore.release(); + if (closed) { + deleteDirectoryQuietly(localLogDir.toFile()); + } + } else { + handleFetchException(request, t, retryCount, startTime); + } + } + } + + private void handleFetchException( + RemoteLogDownloadRequest request, Throwable throwable, int retryCount, long startTime) { + if (closed || request.future.isCancelled()) { + if (!request.future.isDone()) { 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); + prefetchSemaphore.release(); + if (closed) { + deleteDirectoryQuietly(localLogDir.toFile()); + } + return; + } + + LOG.error( + "Failed to download remote log segment file {} for table bucket {}.", + request.getFsPathAndFileName().getFileName(), + request.getTableBucket(), + ExceptionUtils.stripExecutionException(throwable)); + scannerMetricGroup.remoteFetchErrorCount().inc(); + if (retryCount >= 1) { + long backoffMs = RETRY_BACKOFF.backoff(maxRetryCount - retryCount); + LOG.warn( + "Retrying download of remote log segment file {} for table bucket {} " + + "in {} ms (retry {}/{}).", + request.getFsPathAndFileName().getFileName(), + request.getTableBucket(), + backoffMs, + maxRetryCount - retryCount + 1, + maxRetryCount); + try { + Thread.sleep(backoffMs); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + prefetchSemaphore.release(); + request.future.completeExceptionally( + new IOException( + "Interrupted while retrying download of remote log segment file " + + request.getFsPathAndFileName().getFileName(), + e)); + return; + } + downloadRemoteLog(request, retryCount - 1, startTime); + } else { + prefetchSemaphore.release(); + request.future.completeExceptionally( + new IOException( + String.format( + "Failed to download remote log segment file %s, retry count %d", + request.getFsPathAndFileName().getFileName(), maxRetryCount), + throwable)); } } diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchBufferTest.java b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchBufferTest.java index 0ae6697176..d0b5209423 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchBufferTest.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchBufferTest.java @@ -17,6 +17,7 @@ package org.apache.fluss.client.table.scanner.log; +import org.apache.fluss.exception.FetchException; import org.apache.fluss.exception.WakeupException; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.record.LogRecordReadContext; @@ -26,6 +27,7 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import java.io.IOException; import java.time.Duration; import java.util.Arrays; import java.util.Collections; @@ -46,6 +48,7 @@ import static org.apache.fluss.record.TestData.TEST_SCHEMA_GETTER; import static org.apache.fluss.testutils.DataTestUtils.genMemoryLogRecordsByObject; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Test for {@link LogFetchBuffer}. */ public class LogFetchBufferTest { @@ -259,7 +262,7 @@ void testPendFetches() throws Exception { Future signal = service.submit(() -> await(logFetchBuffer, Duration.ofSeconds(1))); - logFetchBuffer.tryComplete(pending1.tableBucket()); + logFetchBuffer.tryComplete(pending1.tableBucket(), null); // nothing happen, as pending1 is not completed assertThat(logFetchBuffer.isEmpty()).isTrue(); // no condition signal @@ -267,7 +270,7 @@ void testPendFetches() throws Exception { signal = service.submit(() -> await(logFetchBuffer, Duration.ofMinutes(1))); completed1.set(true); - logFetchBuffer.tryComplete(pending1.tableBucket()); + logFetchBuffer.tryComplete(pending1.tableBucket(), null); assertThat(signal.get()).isTrue(); assertThat(logFetchBuffer.isEmpty()).isFalse(); assertThat(logFetchBuffer.poll().tableBucket).isEqualTo(tableBucket1); @@ -277,11 +280,11 @@ void testPendFetches() throws Exception { signal = service.submit(() -> await(logFetchBuffer, Duration.ofMinutes(1))); completed2.set(true); - logFetchBuffer.tryComplete(pending2.tableBucket()); + logFetchBuffer.tryComplete(pending2.tableBucket(), null); assertThat(signal.get()).isTrue(); assertThat(logFetchBuffer.isEmpty()).isFalse(); - logFetchBuffer.tryComplete(pending3.tableBucket()); - logFetchBuffer.tryComplete(pending4.tableBucket()); + logFetchBuffer.tryComplete(pending3.tableBucket(), null); + logFetchBuffer.tryComplete(pending4.tableBucket(), null); assertThat(logFetchBuffer.poll().tableBucket).isEqualTo(tableBucket2); assertThat(logFetchBuffer.poll().tableBucket).isEqualTo(tableBucket3); assertThat(logFetchBuffer.poll().tableBucket).isEqualTo(tableBucket3); @@ -289,6 +292,36 @@ void testPendFetches() throws Exception { } } + @Test + void testFetchException() throws Exception { + ExecutorService service = Executors.newSingleThreadExecutor(); + try (LogFetchBuffer logFetchBuffer = new LogFetchBuffer()) { + AtomicBoolean completed = new AtomicBoolean(false); + PendingFetch pendingFetch = makePendingFetch(tableBucket1, completed); + + logFetchBuffer.tryComplete(pendingFetch.tableBucket(), null); + assertThat(logFetchBuffer.isEmpty()).isTrue(); + logFetchBuffer.pend(pendingFetch); + assertThat(logFetchBuffer.isEmpty()).isTrue(); + + Future signal = + service.submit(() -> await(logFetchBuffer, Duration.ofMinutes(1))); + completed.set(true); + logFetchBuffer.tryComplete( + pendingFetch.tableBucket(), new IOException("Test fetch exception")); + assertThat(signal.get()).isTrue(); + assertThat(logFetchBuffer.isEmpty()).isFalse(); + assertThatThrownBy(logFetchBuffer::poll) + .isExactlyInstanceOf(FetchException.class) + .hasMessageContaining("Test fetch exception"); + assertThatThrownBy(logFetchBuffer::peek) + .isExactlyInstanceOf(FetchException.class) + .hasMessageContaining("Test fetch exception"); + } finally { + service.shutdownNow(); + } + } + private boolean await(LogFetchBuffer buffer, Duration waitTime) throws InterruptedException { return buffer.awaitNotEmpty(System.nanoTime() + waitTime.toNanos()); } 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 100853f796..53cfb4ba96 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 @@ -26,12 +26,16 @@ import org.apache.fluss.cluster.Cluster; import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; +import org.apache.fluss.exception.FetchException; import org.apache.fluss.exception.NotLeaderOrFollowerException; +import org.apache.fluss.fs.FsPath; 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.remote.RemoteLogFetchInfo; +import org.apache.fluss.remote.RemoteLogSegment; import org.apache.fluss.rpc.entity.FetchLogResultForBucket; import org.apache.fluss.rpc.messages.FetchLogRequest; import org.apache.fluss.rpc.messages.FetchLogResponse; @@ -45,12 +49,16 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import java.io.IOException; +import java.nio.file.Path; +import java.time.Duration; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -65,7 +73,9 @@ import static org.apache.fluss.record.TestData.DATA1_TABLE_PATH; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getFetchLogData; import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeFetchLogResponse; +import static org.apache.fluss.testutils.common.CommonTestUtils.retry; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** UT Test for {@link LogFetcher}. */ public class LogFetcherTest { @@ -88,18 +98,26 @@ public void tearDown() { } private LogFetcher createLogFetcher(Configuration conf) { - LogScannerStatus logScannerStatus = initializeLogScannerStatus(); + return createLogFetcher(conf, metadataUpdater, new RemoteFileDownloader(1)); + } + + private LogFetcher createLogFetcher( + Configuration conf, + TestingMetadataUpdater updater, + RemoteFileDownloader remoteFileDownloader) { LogFetcher fetcher = new LogFetcher( "default-fetcher", - logScannerStatus, + initializeLogScannerStatus(), conf, - metadataUpdater, + updater, TestingScannerMetricGroup.newInstance(), - new RemoteFileDownloader(1), + remoteFileDownloader, LogRecordReadContext.SchemaResolution.TARGET); fetcher.registerTable( - new TableScanSpec(DATA1_TABLE_INFO, null, null), createSchemaGetter(conf)); + new TableScanSpec(DATA1_TABLE_INFO, null, null), + new TestingClientSchemaGetter( + DATA1_TABLE_PATH, new SchemaInfo(DATA1_SCHEMA, 0), updater, conf)); return fetcher; } @@ -223,6 +241,35 @@ void testPrepareFetchLogRequestWithReadPreference() throws Exception { } } + @Test + void throwExceptionWhenRemoteDownloadFails() throws Exception { + TestingMetadataUpdater localMetadataUpdater = + initializeMetadataUpdater(new RemoteFetchTabletServerGateway()); + + try (RemoteFileDownloader failingDownloader = + new RemoteFileDownloader(1) { + @Override + protected long downloadFile(Path targetFilePath, FsPath remoteFilePath) + throws IOException { + throw new IOException("Simulated remote download failure"); + } + }; + LogFetcher fetcher = + createLogFetcher( + new Configuration(), localMetadataUpdater, failingDownloader)) { + + Map requestMap = + fetcher.prepareFetchLogRequests(Collections.singletonList(tb1)); + fetcher.sendFetchRequest(1, requestMap.get(1)); + retry(Duration.ofSeconds(30), () -> assertThat(fetcher.hasAvailableFetches()).isTrue()); + // collectFetch should throw FetchException due to download failure + assertThatThrownBy(fetcher::collectFetch) + .isInstanceOf(FetchException.class) + .rootCause() + .hasMessageContaining("Simulated remote download failure"); + } + } + private LogScannerStatus initializeLogScannerStatus() { Map scanBucketAndOffsets = new HashMap<>(); scanBucketAndOffsets.put(tb1, 0L); @@ -231,6 +278,43 @@ private LogScannerStatus initializeLogScannerStatus() { return status; } + private static class RemoteFetchTabletServerGateway extends TestTabletServerGateway { + + public RemoteFetchTabletServerGateway() { + super(false, Collections.emptySet()); + } + + @Override + public CompletableFuture fetchLog(FetchLogRequest request) { + Map fetchLogData = getFetchLogData(request); + Map resultForBucketMap = new HashMap<>(); + fetchLogData.forEach( + (tableBucket, fetchReqInfo) -> { + RemoteLogSegment segment = + RemoteLogSegment.Builder.builder() + .tableBucket(tableBucket) + .physicalTablePath(PhysicalTablePath.of(DATA1_TABLE_PATH)) + .remoteLogSegmentId(UUID.randomUUID()) + .remoteLogStartOffset(fetchReqInfo.getFetchOffset()) + .remoteLogEndOffset(fetchReqInfo.getFetchOffset() + 100) + .maxTimestamp(1000L) + .segmentSizeInBytes(1024) + .build(); + RemoteLogFetchInfo remoteLogFetchInfo = + new RemoteLogFetchInfo( + "/tmp/test-tablet-dir", + null, + Collections.singletonList(segment), + 0); + resultForBucketMap.put( + tableBucket, + FetchLogResultForBucket.remote( + tableBucket, remoteLogFetchInfo, 100L)); + }); + return CompletableFuture.completedFuture(makeFetchLogResponse(resultForBucketMap)); + } + } + private static class TestingTabletServerGateway extends TestTabletServerGateway { public 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 8e2fd892e8..aec4252b86 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 @@ -55,6 +55,7 @@ import java.util.Map; import java.util.Set; import java.util.UUID; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -596,6 +597,142 @@ protected long downloadFile(Path targetFilePath, FsPath remoteFilePath) } } + @Test + void testDiscardDoesNotReportDownloadFailure() { + CompletableFuture logFileFuture = new CompletableFuture<>(); + AtomicInteger completionCount = new AtomicInteger(); + RemoteLogDownloadFuture remoteLogDownloadFuture = + new RemoteLogDownloadFuture( + logFileFuture, () -> {}, () -> logFileFuture.cancel(false)); + remoteLogDownloadFuture.whenComplete(ignored -> completionCount.incrementAndGet()); + + remoteLogDownloadFuture.discard(); + + assertThat(logFileFuture).isCancelled(); + assertThat(completionCount).hasValue(0); + } + + @Test + void testFetchException() { + conf.set(ConfigOptions.CLIENT_SCANNER_REMOTE_LOG_PREFETCH_NUM, 1); + RemoteFileDownloader remoteFileDownloader = + new RemoteFileDownloader(1) { + @Override + protected long downloadFile(Path targetFilePath, FsPath remoteFilePath) + throws IOException { + throw new IOException("Test fetch exception"); + } + }; + RemoteLogDownloader remoteLogDownloader = + new RemoteLogDownloader( + DATA1_TABLE_PATH.toString(), + conf, + remoteFileDownloader, + scannerMetricGroup, + 10L); + try { + remoteLogDownloader.start(); + + TableBucket tableBucket = new TableBucket(DATA1_TABLE_ID, 0); + RemoteLogSegment nonExistLogSegment = + RemoteLogSegment.Builder.builder() + .tableBucket(tableBucket) + .physicalTablePath(DATA1_PHYSICAL_TABLE_PATH) + .remoteLogSegmentId(UUID.randomUUID()) + .remoteLogStartOffset(1) + .remoteLogEndOffset(2) + .maxTimestamp(2) + .segmentSizeInBytes(Integer.MAX_VALUE) + .build(); + + RemoteLogDownloadFuture remoteLogDownloadFuture = + remoteLogDownloader.requestRemoteLog(remoteLogDir, nonExistLogSegment); + retry( + Duration.ofMinutes(1), + () -> assertThat(remoteLogDownloadFuture.isDone()).isTrue()); + assertThatThrownBy(() -> remoteLogDownloadFuture.getFileLogRecords(1)) + .cause() + .isInstanceOf(IOException.class) + .hasMessageContaining( + String.format( + "Failed to download remote log segment file %s, retry count %d", + RemoteLogDownloader.getFsPathAndFileName( + remoteLogDir, nonExistLogSegment) + .getFileName(), + conf.getInt( + ConfigOptions + .CLIENT_SCANNER_REMOTE_LOG_FETCH_MAX_RETRIES))) + .rootCause() + .hasMessageContaining("Test fetch exception"); + assertThat(scannerMetricGroup.remoteFetchRequestCount().getCount()).isEqualTo(6); + + RemoteLogSegment secondLogSegment = + RemoteLogSegment.Builder.builder() + .tableBucket(tableBucket) + .physicalTablePath(DATA1_PHYSICAL_TABLE_PATH) + .remoteLogSegmentId(UUID.randomUUID()) + .remoteLogStartOffset(3) + .remoteLogEndOffset(4) + .maxTimestamp(4) + .segmentSizeInBytes(Integer.MAX_VALUE) + .build(); + RemoteLogDownloadFuture secondDownloadFuture = + remoteLogDownloader.requestRemoteLog(remoteLogDir, secondLogSegment); + retry(Duration.ofMinutes(1), () -> assertThat(secondDownloadFuture.isDone()).isTrue()); + assertThat(scannerMetricGroup.remoteFetchRequestCount().getCount()).isEqualTo(12); + } finally { + IOUtils.closeQuietly(remoteLogDownloader); + IOUtils.closeQuietly(remoteFileDownloader); + } + } + + @Test + void testConfigurableMaxRetryCount() { + conf.set(ConfigOptions.CLIENT_SCANNER_REMOTE_LOG_PREFETCH_NUM, 1); + conf.set(ConfigOptions.CLIENT_SCANNER_REMOTE_LOG_FETCH_MAX_RETRIES, 0); + RemoteFileDownloader remoteFileDownloader = + new RemoteFileDownloader(1) { + @Override + protected long downloadFile(Path targetFilePath, FsPath remoteFilePath) + throws IOException { + throw new IOException("Test fetch exception"); + } + }; + RemoteLogDownloader remoteLogDownloader = + new RemoteLogDownloader( + DATA1_TABLE_PATH.toString(), + conf, + remoteFileDownloader, + scannerMetricGroup, + 10L); + try { + remoteLogDownloader.start(); + TableBucket tableBucket = new TableBucket(DATA1_TABLE_ID, 0); + RemoteLogSegment segment = + RemoteLogSegment.Builder.builder() + .tableBucket(tableBucket) + .physicalTablePath(DATA1_PHYSICAL_TABLE_PATH) + .remoteLogSegmentId(UUID.randomUUID()) + .remoteLogStartOffset(1) + .remoteLogEndOffset(2) + .maxTimestamp(2) + .segmentSizeInBytes(Integer.MAX_VALUE) + .build(); + + RemoteLogDownloadFuture future = + remoteLogDownloader.requestRemoteLog(remoteLogDir, segment); + retry(Duration.ofMinutes(1), () -> assertThat(future.isDone()).isTrue()); + assertThat(scannerMetricGroup.remoteFetchRequestCount().getCount()).isEqualTo(1); + assertThatThrownBy(() -> future.getFileLogRecords(1)) + .cause() + .isInstanceOf(IOException.class) + .hasMessageContaining("retry count 0"); + } finally { + IOUtils.closeQuietly(remoteLogDownloader); + IOUtils.closeQuietly(remoteFileDownloader); + } + } + private static class BlockingRemoteFileDownloader extends RemoteFileDownloader { private final CountDownLatch duplicateDownloadCopyStarted; private final CountDownLatch continueDuplicateDownload; diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java index d7f0afd796..50fcd922a9 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java @@ -1615,6 +1615,16 @@ public class ConfigOptions { "The number of remote log segments to keep in local temp file for LogScanner, " + "which download from remote storage. The default setting is 4."); + public static final ConfigOption CLIENT_SCANNER_REMOTE_LOG_FETCH_MAX_RETRIES = + key("client.scanner.remote-log.fetch.max-retries") + .intType() + .defaultValue(5) + .withDescription( + "The maximum number of retries for downloading a remote log segment file. " + + "Each retry is delayed by an exponential backoff starting from " + + "100ms and doubling up to a maximum of 5s. " + + "The default setting is 5."); + public static final ConfigOption CLIENT_SCANNER_LOG_READ_PREFERENCE = key("client.scanner.log.read-preference") .enumType(FetchLogReadPreference.class)