[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.

[SbloodyS]

The legacy fallback is still unbounded on the worker side

The maxFrameSize check only protects the receiving API process. It does not make the legacy RPC memory-bounded on an old worker.

When the chunk RPC is unavailable during a rolling upgrade, the current flow is:

1. streamWholeLog() falls back to writeLocalLegacy().
2. localLogClient.getWholeLog() invokes the old worker RPC.
3. The old worker reads the entire file into a ByteArrayOutputStream.
4. It creates another full byte[] with toByteArray().
5. The RPC layer JSON/base64-serializes the complete response.
6. Only after all of those allocations does the API-side TransporterDecoder receive a frame and enforce maxFrameSize.

Therefore, a sufficiently large log can still OOM the old worker before the decoder has anything to reject. JSON/base64 serialization also adds substantial peak-memory amplification beyond the raw file size.

This means the previous request to either reject the unsupported old-worker path explicitly or make it enforceably bounded has not been addressed. A receiver-side frame limit cannot provide a sender-side memory bound.

Please avoid invoking the whole-file RPC when the chunk method is unavailable. If remote log storage can provide the file, stream it from there; otherwise return an explicit “worker upgrade required for large log download” error. If compatibility for small logs must be preserved, it needs a mechanism that can establish a safe size before requesting the full payload—calling the legacy RPC first is inherently unbounded.

Please also add a rolling-upgrade regression test that verifies an old worker is never asked for the whole-file payload on the large-log path.

[xmg333]

Thanks for the review. The download path no longer falls back to the legacy whole-file RPC, which prevents OOM on the worker.

Changes

• Removed LogClientDelegate.writeLocalLegacy() and getWholeLogBytes().
• Removed the API-side LocalLogClient.getWholeLog() path.
• Removed the byte[]-based RemoteLogClient.getWholeLog() and LoggerService.getLogBytes().
• Removed the unused LogUtils.getFileContentBytesFromRemote().
• There is now no download path that invokes getTaskInstanceWholeLogFileBytes.

The RPC interface and the worker/master-side LogServiceImpl implementation are kept unchanged, so an old API server keeps working against a new worker/master — the reverse direction of the rolling upgrade.

When the chunk RPC is not available, the first-chunk failure now falls back directly to RemoteLogClient.streamWholeLog(), which streams the archived log in bounded chunks.

If remote log storage is also unavailable, the download fails instead of trying the legacy RPC again. The error message distinguishes between:

• the worker not supporting the chunk RPC (it answered but could not dispatch the method), in which case it reports that the worker needs to be upgraded;
• the worker being unreachable (no answer at all — connect refused / timeout), which is reported as a reachability problem rather than a version problem;
• the worker returning a normal non-success response, which is reported as-is.

The remote-storage error is preserved as the cause.

I also added RollingUpgradeLogStreamingIntegrationTest to cover this case. The test uses an embedded Netty RPC server to simulate an old worker. The chunk RPC fails, while the legacy whole-file RPC is still available and returns a 9 MB payload. The legacy RPC invocation count is tracked with an AtomicInteger. Note the assertion is not vacuous: if a future change reintroduced the whole-file call, the stub would return a successful 9 MB response and both the counter and the content assertions would fail.

The test verifies two cases:

• Remote storage succeeds: the download is served from remote storage and the legacy whole-file RPC is never called.
• Remote storage also fails: the download returns the worker-upgrade error and the legacy whole-file RPC is still never called.

So the old worker is no longer asked to construct the whole log payload during the rolling-upgrade download path.

I also removed the small-log compatibility fallback. There is currently no way to determine the log size safely before invoking the legacy RPC, so keeping that fallback would still leave the sender-side allocation unbounded. Log viewing is unaffected because it continues to use the line-bounded API.

The changes are covered by the new integration test and the LogClientDelegateTest cases for first-chunk failures, unsupported RPCs, worker connectivity failures, remote-storage failures, empty logs, and mid-stream errors.

Verification

40 unit/integration tests green on dolphinscheduler-api (JDK 8) plus LogUtilsTest in dolphinscheduler-common; ./mvnw spotless:check passes.

[SbloodyS]

1. ** Keep the remote log file stable throughout streaming**

   [Code evidence] RemoteLogClient.java:99 releases the lock before streaming the file to the HTTP response. A concurrent download or log-view request can then truncate and rewrite the same file through the S3/ABS handler. The active download can hit premature EOF and fail even though the archived log is valid. A slow client extends this race window across the entire transfer.

   Please use a separate temporary file per download, or download to a temporary file and atomically replace the cache, so existing readers retain a stable file.

2. ** Wait for async completion before asserting the response body**

   [Code evidence] LoggerControllerStreamingTest.java:125–131 asserts the response body immediately after the initial MockMvc.perform(). StreamingResponseBody writes on an asynchronous thread, and the initial request does not wait for that write to finish. The assertion can therefore observe an empty response and fail intermittently; it also does not verify the final asynchronous dispatch.

   Please assert asyncStarted(), capture the MvcResult, and check the final response through asyncDispatch(mvcResult).

[xmg333]

Thanks for the review. Both findings are addressed.

1. Keep the remote log file stable during streaming

Implemented the first option: RemoteLogClient.streamWholeLog now creates a private snapshot for each download.

• The cache is copied to a unique temporary file (<archive>.download-<uuid>) under the striped lock, and streaming reads from the snapshot instead of the cache.
• The snapshot is deleted after the transfer, including failure paths. A startup sweep also removes snapshots older than 1 hour.
• This removes the race with a concurrent re-download truncating the cache. The existing short-read check remains as a defensive check.
• The implementation documents the additional disk usage (~2x peak disk space per download).

Added a regression test, streamWholeLog_concurrentCacheRewriteDuringTransfer_activeDownloadUnaffected, which reproduces the concurrent cache rewrite and verifies that the active download still receives the original content. The test fails with the previous implementation, so it covers the actual race.

2. Wait for async completion before asserting the response body

Implemented as suggested. The success test now verifies request().asyncStarted(), captures the MvcResult, and uses asyncDispatch(mvcResult) before asserting the status, headers, and response body.