[Improvement-18459][Common] Stream task log download in bounded chunks to prevent OOM

[xmg333]

Was this PR generated or assisted by AI?

YES. Implementation and tests drafted with assistance from Claude (Anthropic); reviewed by human.

Purpose of the pull request

getFileContentBytesFromLocal read entire files into memory with no size limit. Downloading a large task log caused OOM on the worker.
This PR caps the read at 47 MB and returns a clear error for oversized logs.

Why 47 MB, not 64 MB? The byte[] is JSON-serialized as base64 (~1.33× expansion) before RPC transmission. 47 MB raw → ~63 MB JSON body, staying under the 64 MB maxFrameSize in TransporterDecoder. 64 MB raw would produce ~86 MB body and be rejected by TooLongFrameException.

close #18459

Brief change log

• LogUtils: add MAX_LOG_DOWNLOAD_SIZE = 47 MB; getFileContentBytesFromLocal stops reading once the limit is reached.
• LogServiceImpl: checks file size before reading; returns ERROR with a clear message for oversized logs instead of silently truncating.

Verify this pull request

This change added tests and can be verified as follows:

• LogServiceImplTest: a 48 MB file returns ERROR with message containing "exceeds maximum download size".
• ./mvnw spotless:check passes.

Pull Request Notice

Pull Request Notice

[SbloodyS]

For a log larger than 47 MB:

1. LogServiceImpl#getTaskInstanceWholeLogFileBytes returns ERROR.
2. LogClientDelegate#getWholeLogBytes treats every local error as a reason to call remoteLogClient.getWholeLog(...).
3. RemoteLogClient calls getFileContentBytesFromRemote, which now uses the same capped reader and silently returns only the first 47 MB.

With remote logging enabled, the API can therefore return a successfully downloaded but truncated log. With remote logging disabled or unavailable, it may return an empty log or a generic download error instead of the clear size-limit message.

Please distinguish “local log unavailable” from “log exceeds the supported size,” propagate the latter to the API, and make the reader fail explicitly rather than silently truncating. The regression test should cover the complete LogClientDelegate/API path, not only LogServiceImpl.

Additionally, the linked issue expects large logs to remain downloadable through chunked streaming. This PR rejects them entirely.

[xmg333]

Thanks for the review @SbloodyS . I've reworked the approach based on your feedback. New ** chunked streaming log ** is now completed. Could you confirm if this scope is what you had in mind?

What's new

New RPC: getTaskInstanceLogFileChunk(path, offset, length) reads an 8 MB range via RandomAccessFile. The API loops this RPC and streams
each chunk to the HTTP response via StreamingResponseBody.

Fallback logic (the important part)

The API server runs a chunked loop with an offset counter tracking bytes already written to the response:

streamWholeLog(taskInstance, outputStream):
    if worker not in registry:
        → remote getWholeLogBytes (legacy, unchanged)

    offset = 0
    loop:
        try:
            chunk = localLogClient.getLogChunk(offset, 8MB)
            if chunk.code != SUCCESS:
                if offset == 0:  ← nothing written yet, safe to fallback
                    → remote getWholeLogBytes; return
                else:            ← bytes already streamed, can't restart
                    → throw IOException
            write chunk.bytes; offset += chunk.bytes.length
            if chunk.eof: return
        catch Exception:          ← old worker (method not found), timeout, etc.
            if offset == 0:      ← still safe
                → remote getWholeLogBytes; return
            else:
                → throw IOException

The core invariant: fallback only happens when offset == 0 (nothing written yet). Once bytes have been streamed (offset > 0), there's no
safe way to restart — falling back to getWholeLogBytes would write the whole file from the beginning, duplicating the prefix that's already in
the response. So mid-stream failures throw instead.

Three concrete scenarios:

Scenario	offset	Behavior
Old worker, first chunk RPC fails (method not found)	0	→ fallback to legacy getWholeLogBytes (rolling upgrade safe)
Worker dies mid-stream after writing 24 MB	24 MB	→ throw IOException (client sees truncated download, not corrupted)
Worker offline from the start	0	→ go directly to remote getWholeLogBytes

Legacy path unchanged: getTaskInstanceWholeLogFileBytes and getFileContentBytesFromLocal are untouched from upstream/dev — no silent
truncation, no size cap added.

readFileRange (the new reader) fails explicitly: missing file → IOException, not empty bytes.

Tests cover all three scenarios in LogClientDelegateTest.

[det101]

If remote returns null/empty (archive missing), this still flush()es and the download ends as HTTP 200 with only the log header. Please throw when bytes are absent so a missing remote log is not reported as a successful download.

[xmg333]

Thanks for catching this. Fixed in LogClientDelegate.writeRemoteLegacy: when remoteLogClient.getWholeLog(...) returns null/empty (remote archive missing), it now throws IOException instead of flushing an empty body. The exception propagates through streamWholeLog → StreamingResponseBody and aborts the response, so a missing log is no longer reported as a successful HTTP 200 download.
While reviewing the fix I also found and fixed a related gap: LoggerServiceImpl.checkDownloadLogAuth validated host but not logPath. It's now fixed and would return a clear error before streaming starts.

[SbloodyS]

Keep every fallback path memory-bounded

LogClientDelegate.java:132-202

The chunked path is bounded, but both fallbacks still load the complete log into memory:

• An old worker or first-chunk failure calls localLogClient.getWholeLog(), whose worker-side implementation uses getFileContentBytesFromLocal() and a ByteArrayOutputStream.
• A missing worker calls remoteLogClient.getWholeLog(), which uses getFileContentBytesFromRemote() and then reads the downloaded file into another complete byte[].

Therefore, downloading a large log from an old worker or remote storage can still reproduce the original OOM. Please stream the remote file from disk and either reject the unsupported old-worker path explicitly or otherwise make it enforceably bounded. Add regression coverage for large fallback logs.

Do not execute the whole-file fallback twice after an error

LogClientDelegate.java:141-164

When the first chunk returns a non-success response, writeLocalLegacy() is called inside the outer try. If that fallback throws—for example, the legacy RPC fails and the remote archive is missing—the outer catch still sees offset == 0 and calls writeLocalLegacy() again.

This repeats the legacy and remote requests and can also retry after a fallback has partially written to the response. Please limit the catch to the chunk RPC itself or otherwise let fallback failures propagate without re-entering the fallback.

BTW, the PR title and description still describe a 47 MB cap, while the implementation now uses chunked streaming. Please update them to match the current approach.

[xmg333]

Thanks for the three points — all have been addressed.

1. "Keep every fallback path memory-bounded"

All three paths are now memory-bounded:

• Chunk path: Didn't change.

• Legacy whole-file path: TransporterDecoder now validates bodyLength against maxFrameSize (64 MB by default, configurable through NettyServerConfig / NettyClientConfig) and throws TooLongFrameException before allocating new byte[bodyLength].

  I chose to reject oversized legacy downloads explicitly when this limit is exceeded. writeLocalLegacy detects the condition through isFrameTooLarge(), which walks the full cause chain, and throws IOException("exceeds the maximum legacy download size"). This avoids both silent truncation and falling back to remote storage, which could potentially serve stale data.

  Logs under approximately 47 MB continue to work normally on old workers. The rejection only affects oversized logs on workers that have not yet been upgraded to the chunked RPC.

• Remote path: RemoteLogClient.streamWholeLog downloads the archive to a local file using streaming I/O and then pipes it to the output in 8 KB chunks. There is no whole-file byte[] allocation.

Additionally, NettyClientHandler.exceptionCaught now completes the pending ResponseFuture immediately with the original cause, tracked through a channel AttributeKey. This ensures that TooLongFrameException propagates to the caller immediately instead of waiting for the RPC timeout.

Without this fix, the original exception cause could be lost and replaced by a RemoteTimeoutException with a null cause, making the isFrameTooLarge() check unreachable.

Test coverage:

• testStreamWholeLogLegacyTooLargePropagatesExplicitly
• testStreamWholeLogRemoteFallbackIsChunked
• TransporterDecoderTest — frame-size rejection
• NettyClientHandlerTest — exception propagation

---

2. Do not execute the whole-file fallback twice after an error

streamWholeLog now uses a needFallback flag that is set inside the try/catch. The fallback call (writeLocalLegacy) has been moved outside both blocks.

This ensures that:

• The fallback is executed at most once.
• Exceptions from the fallback propagate directly.
• The fallback exception cannot re-enter the original catch block and trigger a second execution.

This is covered by:

testStreamWholeLogRpcThrowsFallsBackToLegacyThenRemote

The test explicitly verifies:

verify(localLogClient, times(1)).getWholeLog(...)

---

3. PR title and description

Updated to:

Stream task log download in bounded chunks to prevent OOM

[SbloodyS]

NettyClientHandler stores only one opaque request ID in the channel attribute OPAQUE_KEY, while NettyRemotingClient reuses the same channel for concurrent RPC requests.

This creates a race:

1. Request A stores opaque A.
2. Request B stores opaque B on the same channel.
3. A frame/decoder error occurs.
4. exceptionCaught() only completes opaque B.
5. Request A remains pending until its timeout.

There is also a second race: when any request completes successfully, doSendSync() unconditionally clears OPAQUE_KEY, which can erase the opaque ID of another request that is still in flight.

Please avoid using a single channel attribute for pending request tracking. On channel failure, all pending ResponseFutures associated with that channel should be completed with the original exception, or the channel should maintain a proper set/map of in-flight opaque IDs. Please also add a regression test with at least two concurrent requests sharing one channel and a decoder exception.

[xmg333]

Thanks for the review. I spent quite a bit of time debugging the previous OPAQUE_KEY fix. In practice, there were more race conditions around the per-channel opaque set and its lifecycle than I initially expected, especially around channel failures and concurrent request completion. Rather than adding more bookkeeping to handle each case, I ended up replacing that approach entirely.

The new approach is simpler: each ResponseFuture records the channel it was sent on, and FUTURE_TABLE is the only place tracking in-flight requests. When a channel fails or closes, we scan the table and fail the futures belonging to that channel. Futures on other channels are left alone.

The lifecycle is now:

A future stays in FUTURE_TABLE until it completes, and completion always removes it.

This also fixes a few cases that were easy to miss with the previous approach:

• A deserialization/response-processing error can no longer leave a request stuck in the table.
• Timeout, interrupt, and write-failure paths don't need separate cleanup logic.
• There is no longer any channel-attribute initialization or swap handling.

The requested regression test is included: concurrentRequestsSharingChannel_decoderExceptionFailsAll

It sends two concurrent requests over the same real Netty channel, triggers a malformed frame, and verifies that both callers receive the decoder error promptly.

I also added ResponseFutureTest.failAllForChannel_* for channel isolation and already-completed futures.

Other fixes

While testing this, I found a few highly related issues and fixed them as well:

• TransporterDecoder: maxFrameSize now covers the entire message (header + body), using long arithmetic. The default value is shared by the client and server configs.
• Log rotation during download no longer silently truncates. If offset > fileLength, the worker reports the LOG_TRUNCATED. If the file disappears during the read, the existing FileNotFoundException is propagated directly.
• Empty and missing logs are now handled differently. A 0-byte log is a valid empty file, while a missing file fails explicitly. The HTTP response head is written lazily so an early failure can still return JSON.
• Concurrent downloads of the same archive use striped per-path locks, and the read is bounded by the size observed at the start. If the file is replaced during the transfer, the download fails instead of returning potentially inconsistent data.
• The download endpoint now has a request-scoped async timeout. This avoids the servlet's default 30s timeout truncating longer downloads without changing the global timeout.

Verification

There are now 80 tests across the 4 modules, including regression tests for the cases above.