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

# 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