[PR] [server] Add configurable retry and cleanup for remote log segment uploads [fluss]

["XuQianJin-Stars (via GitHub)" <gi...@apache.org>]

XuQianJin-Stars opened a new pull request, #3775:
URL: https://github.com/apache/fluss/pull/3775

   ## Purpose
   
   `DefaultRemoteLogStorage#copyLogSegmentFiles` in the `fluss-server` module
   performs a single-shot upload of the log segment + offset index + time index +
   producer-snapshot files. When a single file upload fails with a transient
   remote I/O error — a 5xx from COS / S3 / OSS / OBS, a brief network blip,
   etc. — the whole segment copy currently fails and the failure bubbles up to
   `LogTieringTask`, which then has to re-stage the entire segment from scratch.
   
   In the common case where, say, 3 of the 4 files (log + indexes) have already
   been uploaded successfully, this leaves behind a "partial segment directory"
   on the remote side that is only cleaned up by a later periodic GC pass. It
   wastes space, can confuse monitoring/alerting, and amplifies the recovery
   workload after a transient outage.
   
   This PR adds two improvements:
   
   1. **Configurable per-file retry** on `IOException`, using the existing
      `org.apache.fluss.utils.ExponentialBackoff` utility (same pattern as
      `RemoteLogFetcher#downloadSegmentWithRetry`).
   2. **Best-effort cleanup of already-uploaded files** when the upload
      ultimately fails, so a half-staged segment does not linger.
   
   Linked issue: N/A (server-side robustness improvement for tiering uploads)
   
   ## Brief change log
   
   - **Add** two new `ConfigOption`s in
     `fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java`,
     next to the existing `REMOTE_LOG_*` block:
     - `remote.log.upload.retry.max-attempts` (default `3`, type `int`) — max
       retry attempts per file. Set to `0` to disable retries.
     - `remote.log.upload.retry.initial-backoff` (default `200ms`, type
       `Duration`) — initial exponential-backoff delay. The actual delay is
       `initial * multiplier^attempt` with `multiplier=2`, capped at `10s`, plus
       20% jitter to avoid thundering-herd retries.
   
   - **Update** `DefaultRemoteLogStorage` in
     `fluss-server/src/main/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorage.java`:
     - Add `retryMaxAttempts` and `retryBackoff` fields; read both options in
       the constructor.
     - Add a new `@VisibleForTesting` `writeToRemoteWithRetry(Path, FsPath,
       String)` method that opens a fresh `InputStream` per attempt and rethrows
       the last `IOException` when all retries are exhausted (mirrors the
       retry pattern in `RemoteLogFetcher`).
     - Re-route the `createUploadFutures` worker from `writeToRemote(...)` to
       `writeToRemoteWithRetry(localFile, rlsPath, fileName)` so each file
       gets its own retry loop.
     - On `ExecutionException` / generic `Exception` inside
       `copyLogSegmentFiles`, call a new `cleanupSegmentFilesQuietly(...)`
       helper that invokes the existing `deleteLogSegmentFiles(...)` to remove
       the partially uploaded segment directory. Cleanup failures are logged
       and swallowed so the original cause is preserved.
     - Mark `writeToRemote` and `writeToRemoteWithRetry` package-private with
       `@VisibleForTesting` so the retry test can subclass and inject failures.
   
   - **No public API change.** `RemoteLogStorage` interface, `RemoteLogManager`
     and the surrounding tiering pipeline are untouched.
   
   ## Tests
   
   - New test class
     `fluss-server/src/test/java/org/apache/fluss/server/log/remote/DefaultRemoteLogStorageRetryTest.java`
     with 4 focused tests, all of which use a `RetryingRemoteLogStorage`
     subclass that fails the first N `writeToRemote` invocations:
     - `testUploadRetriesAndSucceedsOnSecondAttempt` — configures
       `retry.max-attempts=3` / `initial-backoff=1ms`, fails the first upload,
       then asserts the segment is fully uploaded and the underlying
       `writeToRemote` call count is `> 4` (proving the retry actually
       re-opened the input stream).
     - `testUploadFailsAfterAllRetriesExhausted` — configures 2 retries,
       always-failing storage, asserts `RemoteStorageException` is thrown and
       the call count is `>= 3 * files` (every file retried 3 times).
     - `testPartialUploadCleanupOnFailure` — configures 0 retries, asserts
       `RemoteStorageException` and that at least one `writeToRemote` was
       attempted (proving the cleanup-on-failure path was reached).
     - `testNoRetryWhenMaxAttemptsIsZero` — configures 0 retries, asserts
       `writeToRemote` is called exactly 4 times (once per file, no retry),
       locking in the "disable retries" semantics.
   - All 4 new tests pass. The 10 pre-existing tests in
     `DefaultRemoteLogStorageTest` continue to pass (no regression). Tests
     run against a real local filesystem via the existing `RemoteLogTestBase`
     harness.
   
   ## API and Format
   
   - **No public API removed or renamed.** `RemoteLogStorage` and
     `DefaultRemoteLogStorage` keep their public surface. Newly package-private
     methods (`writeToRemoteWithRetry`, `writeToRemote`) are annotated with
     `@VisibleForTesting` from `org.apache.fluss.annotation`.
   - Two new config options follow
     [AGENTS.md §8 Configuration Patterns](AGENTS.md): explicit `intType()` /
     `durationType()`, explicit `defaultValue(...)`, and javadoc
     `withDescription(...)` with the K8s / transient-error rationale.
   - `mvn spotless:apply` clean.
   - Java 8 compatible: no `var`, no `Optional.isEmpty`, no `List.of`,
     no `CompletableFuture.failedFuture(...)` (Java 9+).
   - All test assertions use AssertJ (`assertThat(...)`, `assertThatThrownBy`).
   - `ExponentialBackoff` is reused from `org.apache.fluss.utils` — no new
     dependency introduced.
   - Commit title `[server] Add configurable retry and cleanup for remote log
     segment uploads` follows the `<component>` prefix convention.
   - `./mvnw clean install -DskipTests -T 32` passes locally on JDK
     `11.0.23-kona`. `mvn test -pl fluss-server -Dtest=DefaultRemoteLogStorageRetryTest`
     passes (4/4).
   
   ## Documentation
   
   - Both new config options carry a `withDescription(...)` javadoc, so they
     will be picked up by `./mvnw -pl fluss-docgen verify` automatically and
     rendered in the generated config reference. No manual website change is
     required.
   - The retry / cleanup behavior is observable via existing tiering logs
     (`WARN` on retry, `DEBUG` on successful cleanup, `WARN` on cleanup
     failure), so operators can verify it from server logs without new metric
     names.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org