From 10bbbf43cb5fec3b9b9257c4b84c465fed4728b1 Mon Sep 17 00:00:00 2001 From: forwardxu Date: Sun, 26 Jul 2026 15:36:24 +0800 Subject: [PATCH] [server] Add configurable retry and cleanup for remote log segment uploads --- .../apache/fluss/config/ConfigOptions.java | 19 ++ .../log/remote/DefaultRemoteLogStorage.java | 93 +++++++- .../DefaultRemoteLogStorageRetryTest.java | 202 ++++++++++++++++++ 3 files changed, 310 insertions(+), 4 deletions(-) create mode 100644 fluss-server/src/test/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorageRetryTest.java 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 233902cb391..01f1bccddd2 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 @@ -1121,6 +1121,25 @@ public class ConfigOptions { + "remote log file can be data file, index file and remote log metadata file. " + "This option is deprecated. Please use server.io-pool.size instead."); + public static final ConfigOption REMOTE_LOG_UPLOAD_RETRY_MAX_ATTEMPTS = + key("remote.log.upload.retry.max-attempts") + .intType() + .defaultValue(3) + .withDescription( + "The maximum number of retry attempts for uploading a single file to " + + "remote storage. Retries help tolerate transient I/O errors " + + "(e.g., temporary 5xx from object storage). Set to 0 to disable " + + "retries."); + + public static final ConfigOption REMOTE_LOG_UPLOAD_RETRY_INITIAL_BACKOFF = + key("remote.log.upload.retry.initial-backoff") + .durationType() + .defaultValue(Duration.ofMillis(200)) + .withDescription( + "The initial backoff delay (in milliseconds) before the first retry " + + "of a failed file upload to remote storage. The delay increases " + + "exponentially with subsequent retries."); + // ------------------------------------------------------------------------ // Netty Settings // ------------------------------------------------------------------------ diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorage.java b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorage.java index 56e8e240915..d9bb1651ae8 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorage.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorage.java @@ -17,6 +17,7 @@ package org.apache.fluss.server.log.remote; +import org.apache.fluss.annotation.VisibleForTesting; import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.exception.RemoteStorageException; @@ -29,6 +30,7 @@ import org.apache.fluss.remote.RemoteLogSegment; import org.apache.fluss.utils.CloseableRegistry; import org.apache.fluss.utils.ExceptionUtils; +import org.apache.fluss.utils.ExponentialBackoff; import org.apache.fluss.utils.FlussPaths; import org.apache.fluss.utils.IOUtils; import org.apache.fluss.utils.concurrent.FutureUtils; @@ -66,10 +68,16 @@ public class DefaultRemoteLogStorage implements RemoteLogStorage { private static final int READ_BUFFER_SIZE = 16 * 1024; + private static final int RETRY_BACKOFF_MULTIPLIER = 2; + private static final long RETRY_BACKOFF_MAX_MS = 10_000; + private static final double RETRY_BACKOFF_JITTER = 0.2; + private final FsPath remoteLogDir; private final FileSystem fileSystem; private final ExecutorService ioExecutor; private final int writeBufferSize; + private final int retryMaxAttempts; + private final ExponentialBackoff retryBackoff; public DefaultRemoteLogStorage(Configuration conf, ExecutorService ioExecutor) throws IOException { @@ -77,6 +85,13 @@ public DefaultRemoteLogStorage(Configuration conf, ExecutorService ioExecutor) this.fileSystem = remoteLogDir.getFileSystem(); this.writeBufferSize = (int) conf.get(ConfigOptions.REMOTE_FS_WRITE_BUFFER_SIZE).getBytes(); this.ioExecutor = ioExecutor; + this.retryMaxAttempts = conf.get(ConfigOptions.REMOTE_LOG_UPLOAD_RETRY_MAX_ATTEMPTS); + this.retryBackoff = + new ExponentialBackoff( + conf.get(ConfigOptions.REMOTE_LOG_UPLOAD_RETRY_INITIAL_BACKOFF).toMillis(), + RETRY_BACKOFF_MULTIPLIER, + RETRY_BACKOFF_MAX_MS, + RETRY_BACKOFF_JITTER); } @Override @@ -106,6 +121,8 @@ public void copyLogSegmentFiles( } catch (ExecutionException e) { Throwable throwable = ExceptionUtils.stripExecutionException(e); throwable = ExceptionUtils.stripException(throwable, RuntimeException.class); + // Clean up any partially uploaded segment files to avoid leaving orphaned data + cleanupSegmentFilesQuietly(remoteLogSegment); throw new RemoteStorageException( "Failed to copy log segment and indexes to remote dir for path: " + remoteLogSegment, @@ -117,6 +134,7 @@ public void copyLogSegmentFiles( + remoteLogSegment, e); } catch (Exception e) { + cleanupSegmentFilesQuietly(remoteLogSegment); throw new RemoteStorageException( "Failed to copy log segment and indexes to remote for path: " + remoteLogSegment, @@ -124,6 +142,22 @@ public void copyLogSegmentFiles( } } + /** + * Attempts to delete already-uploaded segment files after a failed upload. Any errors during + * cleanup are logged but not propagated, so the original failure cause is preserved. + */ + private void cleanupSegmentFilesQuietly(RemoteLogSegment remoteLogSegment) { + try { + deleteLogSegmentFiles(remoteLogSegment); + LOG.debug("Cleaned up partially uploaded segment files for: {}", remoteLogSegment); + } catch (Exception cleanupException) { + LOG.warn( + "Failed to clean up partially uploaded segment files for: {}", + remoteLogSegment, + cleanupException); + } + } + /** * Delete log segments from remote path. Currently, these files need to be deleted: * @@ -286,8 +320,8 @@ private List> createUploadFutures( CompletableFuture.runAsync( ThrowingRunnable.unchecked( () -> - writeToRemote( - Files.newInputStream(localFile), + writeToRemoteWithRetry( + localFile, rlsPath, localFile.getFileName().toString())), ioExecutor); @@ -296,6 +330,55 @@ private List> createUploadFutures( return list; } + /** + * Writes a local file to remote storage with configurable retry on {@link IOException}. + * + *

On each retry attempt, a fresh input stream is opened from the local file. If all retry + * attempts are exhausted, the last exception is rethrown. + */ + @VisibleForTesting + void writeToRemoteWithRetry(Path localFile, FsPath remoteDir, String remoteFileName) + throws IOException { + IOException lastException = null; + for (int attempt = 0; attempt <= retryMaxAttempts; attempt++) { + try { + writeToRemote(Files.newInputStream(localFile), remoteDir, remoteFileName); + return; + } catch (IOException e) { + lastException = e; + if (attempt == retryMaxAttempts) { + break; + } + long retryDelayMs = retryBackoff.backoff(attempt); + LOG.warn( + "Failed to upload file {} to remote path {} on attempt {}/{}. " + + "Retry after {} ms.", + remoteFileName, + remoteDir, + attempt + 1, + retryMaxAttempts + 1, + retryDelayMs, + e); + try { + Thread.sleep(retryDelayMs); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw new IOException( + "Interrupted while retrying file upload: " + remoteFileName, ie); + } + } + } + throw new IOException( + "Failed to upload file " + + remoteFileName + + " to remote path " + + remoteDir + + " after " + + (retryMaxAttempts + 1) + + " attempts", + lastException); + } + /** * Write input stream to remote. * @@ -304,8 +387,10 @@ private List> createUploadFutures( * @param remoteDir remote dir * @return remote file path including file name */ - private @Nullable FsPath writeToRemote( - InputStream inputStream, FsPath remoteDir, String remoteFileName) throws IOException { + @VisibleForTesting + @Nullable + FsPath writeToRemote(InputStream inputStream, FsPath remoteDir, String remoteFileName) + throws IOException { try (CloseableRegistry closeableRegistry = new CloseableRegistry()) { final byte[] buffer = new byte[READ_BUFFER_SIZE]; closeableRegistry.registerCloseable(inputStream); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorageRetryTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorageRetryTest.java new file mode 100644 index 00000000000..07d51039ab8 --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorageRetryTest.java @@ -0,0 +1,202 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.server.log.remote; + +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.exception.RemoteStorageException; +import org.apache.fluss.fs.FsPath; +import org.apache.fluss.remote.RemoteLogSegment; +import org.apache.fluss.server.log.LogTablet; +import org.apache.fluss.utils.FlussPaths; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import javax.annotation.Nullable; + +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.time.Duration; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for the retry and cleanup behavior of {@link DefaultRemoteLogStorage}. */ +class DefaultRemoteLogStorageRetryTest extends RemoteLogTestBase { + + private ExecutorService ioExecutor; + + @BeforeEach + @Override + public void setup() throws Exception { + super.setup(); + ioExecutor = Executors.newSingleThreadExecutor(); + } + + @AfterEach + public void teardown() throws Exception { + if (ioExecutor != null) { + ioExecutor.shutdown(); + } + } + + @Test + void testUploadRetriesAndSucceedsOnSecondAttempt() throws Exception { + // Configure with 3 retries and minimal backoff + conf.set(ConfigOptions.REMOTE_LOG_UPLOAD_RETRY_MAX_ATTEMPTS, 3); + conf.set(ConfigOptions.REMOTE_LOG_UPLOAD_RETRY_INITIAL_BACKOFF, Duration.ofMillis(1)); + + // A storage that fails the first writeToRemote call, then succeeds + AtomicInteger writeAttempts = new AtomicInteger(0); + RetryingRemoteLogStorage storage = + new RetryingRemoteLogStorage(conf, ioExecutor, writeAttempts, 1); + + LogTablet logTablet = makeLogTabletAndAddSegments(false); + RemoteLogSegment remoteLogSegment = copyLogSegmentToRemote(logTablet, storage, 0); + + // Verify the segment was uploaded successfully + File remoteLogDir = getTestingRemoteLogSegmentDir(remoteLogSegment, storage); + assertThat(remoteLogDir.exists()).isTrue(); + assertThat(remoteLogDir.listFiles()).isNotNull().hasSize(4); + + // Verify that writeToRemote was called more than once (proving retry happened) + assertThat(writeAttempts.get()).isGreaterThan(4); // 4 files, at least one retried + + storage.close(); + } + + @Test + void testUploadFailsAfterAllRetriesExhausted() throws Exception { + // Configure with 2 retries and minimal backoff + conf.set(ConfigOptions.REMOTE_LOG_UPLOAD_RETRY_MAX_ATTEMPTS, 2); + conf.set(ConfigOptions.REMOTE_LOG_UPLOAD_RETRY_INITIAL_BACKOFF, Duration.ofMillis(1)); + + // A storage that always fails + AtomicInteger writeAttempts = new AtomicInteger(0); + RetryingRemoteLogStorage storage = + new RetryingRemoteLogStorage(conf, ioExecutor, writeAttempts, Integer.MAX_VALUE); + + LogTablet logTablet = makeLogTabletAndAddSegments(false); + + // copyLogSegmentToRemote should throw RemoteStorageException + assertThatThrownBy(() -> copyLogSegmentToRemote(logTablet, storage, 0)) + .isInstanceOf(RemoteStorageException.class); + + // Verify that writeToRemote was retried: each file should have been attempted + // (maxAttempts + 1) = 3 times + assertThat(writeAttempts.get()).isGreaterThanOrEqualTo(3); + + storage.close(); + } + + @Test + void testPartialUploadCleanupOnFailure() throws Exception { + // Configure with 0 retries (fail immediately) + conf.set(ConfigOptions.REMOTE_LOG_UPLOAD_RETRY_MAX_ATTEMPTS, 0); + conf.set(ConfigOptions.REMOTE_LOG_UPLOAD_RETRY_INITIAL_BACKOFF, Duration.ofMillis(1)); + + // A storage that fails all writes + AtomicInteger writeAttempts = new AtomicInteger(0); + RetryingRemoteLogStorage storage = + new RetryingRemoteLogStorage(conf, ioExecutor, writeAttempts, Integer.MAX_VALUE); + + LogTablet logTablet = makeLogTabletAndAddSegments(false); + + assertThatThrownBy(() -> copyLogSegmentToRemote(logTablet, storage, 0)) + .isInstanceOf(RemoteStorageException.class); + + // After failure, the cleanup should have deleted any partially uploaded files. + // Since all writes failed, the remote directory may or may not exist, but if it + // does, it should be empty (all files were either never created or cleaned up). + // We verify by checking that writeAttempts > 0 (at least one write was attempted) + assertThat(writeAttempts.get()).isGreaterThan(0); + + storage.close(); + } + + @Test + void testNoRetryWhenMaxAttemptsIsZero() throws Exception { + // Configure with 0 retries + conf.set(ConfigOptions.REMOTE_LOG_UPLOAD_RETRY_MAX_ATTEMPTS, 0); + conf.set(ConfigOptions.REMOTE_LOG_UPLOAD_RETRY_INITIAL_BACKOFF, Duration.ofMillis(1)); + + AtomicInteger writeAttempts = new AtomicInteger(0); + RetryingRemoteLogStorage storage = + new RetryingRemoteLogStorage(conf, ioExecutor, writeAttempts, Integer.MAX_VALUE); + + LogTablet logTablet = makeLogTabletAndAddSegments(false); + + assertThatThrownBy(() -> copyLogSegmentToRemote(logTablet, storage, 0)) + .isInstanceOf(RemoteStorageException.class); + + // With 0 retries, each file should only be attempted once (no retry) + // 4 files (log, offset index, timestamp index, producer snapshot) + assertThat(writeAttempts.get()).isEqualTo(4); + + storage.close(); + } + + private File getTestingRemoteLogSegmentDir( + RemoteLogSegment remoteLogSegment, RemoteLogStorage storage) { + return new File( + FlussPaths.remoteLogSegmentDir( + FlussPaths.remoteLogTabletDir( + storage.getRemoteLogDir(), + remoteLogSegment.physicalTablePath(), + remoteLogSegment.tableBucket()), + remoteLogSegment.remoteLogSegmentId()) + .toString()); + } + + /** + * A {@link DefaultRemoteLogStorage} subclass that injects failures into {@code writeToRemote} + * for the first {@code failFirstNCalls} calls, then delegates to the real implementation. + */ + private static class RetryingRemoteLogStorage extends DefaultRemoteLogStorage { + private final AtomicInteger writeCallCounter; + private final int failFirstNCalls; + + RetryingRemoteLogStorage( + org.apache.fluss.config.Configuration conf, + ExecutorService ioExecutor, + AtomicInteger writeCallCounter, + int failFirstNCalls) + throws IOException { + super(conf, ioExecutor); + this.writeCallCounter = writeCallCounter; + this.failFirstNCalls = failFirstNCalls; + } + + @Nullable + @Override + FsPath writeToRemote(InputStream inputStream, FsPath remoteDir, String remoteFileName) + throws IOException { + int callNum = writeCallCounter.incrementAndGet(); + if (callNum <= failFirstNCalls) { + throw new IOException( + "Simulated failure on call " + callNum + " for file " + remoteFileName); + } + return super.writeToRemote(inputStream, remoteDir, remoteFileName); + } + } +}