Aiven-Open / Aiven-Open/tiered-storage-for-apache-kafka

RemoteStorageManager.copyLogSegmentData() hangs indefinitely on intermittent uploads — never returns, never throws

Abierto
#820 9 comentarios 9 reacciones 0 asignados Ver en GitHub
Lenguaje dominante
Java
Estrellas
239
Forks
59
Métricas de merge de PR
Sin PR fusionados en 30 d

Descripción

# What happened?

`RemoteStorageManager.copyLogSegmentData()` is invoked and emits its standard start log line, but for a specific segment **the matching completion log never fires and no exception is thrown** — the method exits silently, violating its caller contract (which expects either normal return with completion log, or `RemoteStorageException`). Recurring — 12 occurrences in ~5 weeks on a single 8-broker production cluster.

**Reproducer — log evidence on broker 1704, segment id `0PoyJGH-SoyRc2OGewoa1g`:**

Aiven RSM emits the start line from `io.aiven.kafka.tieredstorage.RemoteStorageManager` (line 100 of `core/src/main/java/.../RemoteStorageManager.java` at v1.1.1):

```
Copying log segment data, metadata: RemoteLogSegmentMetadata{remoteLogSegmentId=RemoteLogSegmentId{topicIdPartition=9bzohKFqSZqNJQHOaaOM0A:xxx-65, id=0PoyJGH-SoyRc2OGewoa1g}, startOffset=61311877702, endOffset=61312752040, brokerId=1704, maxTimestampMs=1778625612430, eventTimestampMs=1778625612903, segmentLeaderEpochs={88=61311877702}, segmentSizeInBytes=99870740, customMetadata=Optional.empty, state=COPY_SEGMENT_STARTED, txnIdxEmpty=true} (io.aiven.kafka.tieredstorage.RemoteStorageManager)
```

The matching `Copying log segment data completed successfully, metadata: ...` from the same class **never fires** for this segment. 10+ hours after the start line, no completion. No exception. No DEBUG/INFO/WARN/ERROR line ties back to this segment id.

Independent confirmation from `__remote_log_metadata` (Kafka's internal RLMM topic): `COPY_SEGMENT_STARTED` for `0PoyJ…` at 2026-05-12T22:40:12Z, `COPY_SEGMENT_FINISHED` never recorded for this segment. After exactly +8 hours (matching `retention.ms=28800000`), Kafka's `RLMExpirationTask` retention-DELETEs the orphan metadata. The S3 multipart upload state is never reconciled.

**AWS SDK-layer is NOT where the failure originates.** With `rsm.config.storage.s3.api.call.timeout=300000` / `rsm.config.storage.s3.api.call.attempt.timeout=60000` deployed (each SDK call bounded ≤ 60 s / overall ≤ 5 min):

- `aiven_kafka_tieredstorage_s3_max{metric=upload-part-time}` bounded ≤ 5000 ms throughout the affected window
- `rate(aiven_kafka_tieredstorage_s3_total[5m])` healthy (~1.1 ops/s) on the affected broker, indistinguishable from peers
- AWS CloudWatch `aws_s3_first_byte_latency_average` 57-76 ms; `5xx_errors_sum` negligible

Each `S3Client.createMultipartUpload` / `uploadPart` / `completeMultipartUpload` call returns within its bound. Furthermore, a live JVM thread dump (`ThreadMXBean.dumpAllThreads`) captured ~13 h after the START log shows **zero threads parked in any `io.aiven.kafka.tieredstorage.*` frame and zero threads parked in any AWS SDK frame for the affected partition**. The failure is not a stuck thread — `copyLogSegmentData()` *returned* (silently, without exception, without completion log). The bug is **inside RSM's upload-coordinator code, between SDK calls** — an unhandled exit path that bypasses both the success log and the exception contract.

# What did you expect to happen?

`RemoteStorageManager.copyLogSegmentData()` either returns normally (emits the standard completion log line) or throws `RemoteStorageException` that the caller can handle. Silently returning without completion log and without exception — the observed behavior — is a contract violation. The caller (`RemoteLogManager.RLMCopyTask`) cannot distinguish "succeeded but failed to log" from "failed but failed to throw" from "skipped for an unknown reason," so it cannot reconcile state or re-attempt.

# What else do we need to know?

## Source-level gap — the upload path has no in-progress instrumentation

With `io.aiven.kafka.tieredstorage=DEBUG` enabled cluster-wide on the affected deployment, inspecting v1.1.1 source:

**`RemoteStorageManager.copyLogSegmentData()`** (lines 94-124): one `log.info` at start (line 100), one `log.info` at complete (line 121). **Zero DEBUG or INFO between.** If the delegated `kafkaRsm.copyLogSegmentData(...)` call at line 112 exits abnormally without propagating an exception, the silent-exit is invisible to logs.

**`KafkaRemoteStorageManager.copyLogSegmentData()`** (lines 166-222): three sequential phases — `uploadSegmentLog` / `uploadIndexes` / `uploadManifest`. Each logs `log.debug("Uploaded {X} for {metadata}, size: {bytes}")` **only on success**. Nothing logs on entry to a phase or during a phase.

**`S3UploadOutputStream`** (the SDK caller):

- `log.debug("Create new multipart upload request: {uploadId}")` (line 144) — ONLY if `createMultipartUpload` succeeds
- `log.debug("Completed multipart upload {uploadId}")` (line 182) — ONLY if `completeMultipartUpload` succeeds
- Per-part `uploadPart()` (lines 249-265): **no logging** — called ~20 times for a 100 MB segment with default 5 MB part size, all silent
- `completeUpload()` (lines 210-221): **no entry log**, **no success log**

**Net:** with full `io.aiven.kafka.tieredstorage=DEBUG` enabled, the affected segment produces:

What we see:
- ✓ INFO `Copying log segment data, metadata: ...` (one-time, from line 100)

What's missing:
- ✗ INFO `Copying log segment data completed successfully, ...` (line 121) — never fires
- ✗ Any WARN/ERROR/exception with the segment id or topic-partition
- ✗ Caller-side `Copy failed, cleaning segment {}` from `RemoteLogManager.RLMCopyTask.copyLogSegment()` — never fires (so no `RemoteStorageException` was thrown)
- ✗ Any subsequent `$RLMCopyTask` re-invocation for the partition (silent forever, until operator intervention)

The bug is **fundamentally undebuggable from logs alone** in v1.1.1. We cannot tell whether the silent exit is in `uploadSegmentLog` / `uploadIndexes` / `uploadManifest` (which phase), nor whether it's at `createMultipartUpload` / per-part `uploadPart` / `completeMultipartUpload` (which SDK call).

**Logs**
```
09:40:32.452 Deleted remote log segment RemoteLogSegmentId{topicIdPartition=9bzohKFqSZqNJQHOaaOM0A:xxx-65, id=0PoyJGH-SoyRc2OGewoa1g} (org.apache.kafka.server.log.remote.storage.RemoteLogManager) DEBUG
09:40:32.444 Deleting log segment data for completed successfully RemoteLogSegmentMetadata{remoteLogSegmentId=RemoteLogSegmentId{topicIdPartition=9bzohKFqSZqNJQHOaaOM0A:xxx-65, id=0PoyJGH-SoyRc2OGewoa1g}, startOffset=61311877702, endOffset=61312752040, brokerId=1704, maxTimestampMs=1778625612430, eventTimestampMs=1778625612903, segmentLeaderEpochs={88=61311877702}, segmentSizeInBytes=99870740, customMetadata=Optional.empty, state=COPY_SEGMENT_STARTED, txnIdxEmpty=true} (io.aiven.kafka.tieredstorage.RemoteStorageManager) INFO
09:40:32.352 Deleting log segment data for RemoteLogSegmentMetadata{remoteLogSegmentId=RemoteLogSegmentId{topicIdPartition=9bzohKFqSZqNJQHOaaOM0A:xxx-65, id=0PoyJGH-SoyRc2OGewoa1g}, startOffset=61311877702, endOffset=61312752040, brokerId=1704, maxTimestampMs=1778625612430, eventTimestampMs=1778625612903, segmentLeaderEpochs={88=61311877702}, segmentSizeInBytes=99870740, customMetadata=Optional.empty, state=COPY_SEGMENT_STARTED, txnIdxEmpty=true} (io.aiven.kafka.tieredstorage.RemoteStorageManager) INFO
09:40:32.341 Deleting remote log segment RemoteLogSegmentId{topicIdPartition=9bzohKFqSZqNJQHOaaOM0A:xxx-65, id=0PoyJGH-SoyRc2OGewoa1g} (org.apache.kafka.server.log.remote.storage.RemoteLogManager) DEBUG
09:40:32.341 [RemoteLogManager=1704 partition=9bzohKFqSZqNJQHOaaOM0A:xxx-65] About to delete remote log segment RemoteLogSegmentId{topicIdPartition=9bzohKFqSZqNJQHOaaOM0A:xxx-65, id=0PoyJGH-SoyRc2OGewoa1g} due to retention time 28800000ms breach based on the largest record timestamp in the segment (org.apache.kafka.server.log.remote.storage.RemoteLogManager$RLMExpirationTask) INFO
01:40:12.912 Copying log segment data, metadata: RemoteLogSegmentMetadata{remoteLogSegmentId=RemoteLogSegmentId{topicIdPartition=9bzohKFqSZqNJQHOaaOM0A:xxx-65, id=0PoyJGH-SoyRc2OGewoa1g}, startOffset=61311877702, endOffset=61312752040, brokerId=1704, maxTimestampMs=1778625612430, eventTimestampMs=1778625612903, segmentLeaderEpochs={88=61311877702}, segmentSizeInBytes=99870740, customMetadata=Optional.empty, state=COPY_SEGMENT_STARTED, txnIdxEmpty=true} (io.aiven.kafka.tieredstorage.RemoteStorageManager) INFO
```

## Affected versions

- `io.aiven.kafka.tieredstorage` **v1.1.1**
- AWS SDK Java v2.34.7
- AWS EC2, us-east-1, single-region S3 bucket
- Apache Kafka 4.2.0 (caller of the RSM)

## Knock-on Kafka-side effect

Once `copyLogSegmentData()` returns silently (without completion log and without `RemoteStorageException`), the caller's `$RLMCopyTask` thread completes its invocation normally — but **no further `$RLMCopyTask` invocations fire for the affected partition** until an operator intervenes via leader-swap reassignment. Direct evidence on the affected broker, ~13 h after the START log:

- A live JVM thread dump shows the `kafka-rlm-copy-thread-pool-*` worker threads idle/parked in their pool queue. **Zero threads are stuck in any `io.aiven.kafka.tieredstorage.*` frame.** This refutes the simpler "the call is still in progress" hypothesis.
- Apache Kafka's `RLMExpirationTask` continues to fire normally on the same partition every 30 s (its retention-DELETE for the orphan COPY_SEGMENT_STARTED metadata is visible in `__remote_log_metadata` and in DEBUG logs above).
- The partition's local disk accumulates rolling segments that never get COPY-initiated. The bloat is bounded by `retention.ms` (default 24 h) but causes per-broker disk pressure and operator pages.

The mechanism by which Kafka's `RLMScheduledThreadPool` stops re-scheduling `$RLMCopyTask` for the partition after a no-exception silent return is being investigated separately on the Apache Kafka side (suspected unchecked `Throwable` escaping `RLMTask.run()`, which under `ScheduledThreadPoolExecutor`'s documented contract would suppress all subsequent `scheduleAtFixedRate` invocations of that task). That investigation is downstream of this issue — **the root cause filed here is the contract violation in Aiven RSM that triggers it.** If `copyLogSegmentData()` had thrown `RemoteStorageException` or returned normally with the completion log, the Apache Kafka downstream behavior would be moot.

## Suggested fixes (in order of preference)

1. **Audit every exit path of `RemoteStorageManager.copyLogSegmentData()` and `KafkaRemoteStorageManager.copyLogSegmentData()` against the invariant: each invocation must either reach the completion log line OR throw `RemoteStorageException`** — there must be no third path. Likely candidates for the silent exit: unchecked `Throwable` (e.g., `OutOfMemoryError`, `StackOverflowError`, `NoClassDefFoundError`) escaping mid-upload without being wrapped; uncaught `InterruptedException` clearing the interrupt flag and returning early; finally-block ordering issues that swallow a thrown exception; per-phase try/catch blocks that log-and-return instead of log-and-rethrow. Add an outer try/finally in `RemoteStorageManager.copyLogSegmentData()` that, on any non-normal exit, emits `log.error("Copy aborted abnormally for segment {}", ..., t)` and rethrows as `RemoteStorageException`. The invariant should be testable in unit-tests by asserting that no public method returns without either a recorded completion log or a thrown `RemoteStorageException`.

2. **Add progress logging across the upload path:**
- Pre-phase entry log in `KafkaRemoteStorageManager`: `"Starting phase {segmentLog|indexes|manifest} for metadata: {...}"`
- Per-part log in `S3UploadOutputStream.uploadPart()`: `"Uploading part {partNumber}, uploadId: {uploadId}, size: {actualPartSize}"`
- Entry log in `S3UploadOutputStream.completeUpload()`: `"Issuing completeMultipartUpload, uploadId: {uploadId}, partCount: {completedParts.size()}"`
- These give operators forward visibility into *where* a silent exit occurred even before the invariant audit lands.

3. **Add an internal overall timeout / fail-fast wrapper around `copyLogSegmentData()`** (separate from SDK-level `apiCallTimeout`), throwing `RemoteStorageException` on exceeded. While the live thread dump shows the affected call is not stuck (it has already exited), a defensive timeout would protect against a future regression in which a thread *does* park indefinitely (e.g., if a JVM-level resource exhaustion blocked the SDK timeout enforcement).

4. **Investigate `S3UploadOutputStream`'s state machine** for transitions that could exit abnormally between SDK calls — multipart-upload-id state, per-part-completion accounting in `completedParts`, interaction with the SDK's connection pool, exception-handling around AbortMultipartUpload cleanup.

## Workaround in production

Same-broker leader-swap reassignment (`kafka-reassign-partitions.sh --execute` + `kafka-leader-election.sh --election-type PREFERRED`) clears the wedged state — the new leader's `$RLMCopyTask` thread starts fresh and drains the backlog at the broker's S3 copy quota. Tracked operationally; not a code fix.

## Related upstream Apache Kafka issues (none describes the exact bug)

- [KAFKA-17980](https://issues.apache.org/jira/browse/KAFKA-17980) — `isReady` gate exists and works; this bug is post-`isReady()`
- [KAFKA-19523](https://issues.apache.org/jira/browse/KAFKA-19523) — follower-fetcher path; ours is leader-side `$RLMCopyTask`
- [KAFKA-16105](https://issues.apache.org/jira/browse/KAFKA-16105) — newly-assigned-cache race; already in 4.2.0; this bug recurs *despite* the fix
- [KAFKA-19385](https://issues.apache.org/jira/browse/KAFKA-19385) — read-side stuck consumption; ours is upload/copy-side

Guía de contribución

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.