Fragment large TsBlocks in MPP DataExchange responses - #18645

[P2] Stop fetching fragments when the source is closed or aborted

If close() or abort() runs while a fragment RPC is in flight, this loop still consumes its response and downloads the remaining fragments because it only checks fetchProgress.isFinished(). The source has already returned its reservation to the query pool, so it can allocate the full PublicBAOS(totalLength) and retain accumulated data after cancellation without an accounted reservation. A focused test that closes or aborts during the first RPC of a three-fragment transfer still executes all three RPCs with getBufferRetainedSizeInBytes() == 0. For large blocks this also keeps the exchange worker and connection occupied until the entire range is fetched. Please check the terminal state with appropriate synchronization before each new RPC and again before consuming its response, and discard the pending fetch state when stopping.

[P2] Preserve failure cleanup for errors during fragment assembly

Narrowing the previous catch (Throwable) to catch (Exception) leaves an OutOfMemoryError from the new PublicBAOS(totalLength) allocation uncaught. This allocation happens after client.getDataBlock() returns, so the Thrift client wrapper cannot convert it into a TException. The submitted task then terminates without calling fail(): the executor only logs the error, leaving the source blocked and its reservation held until an outer timeout or cancellation. Using the JVM array-size limit to safely inject an allocation error reproduced zero failure callbacks, a blocked source, and a retained 1,024-byte reservation. Please also notify the fragment of failure and release the reservation on the Error path; failing immediately is sufficient and does not require retrying an OOM.

[P3] Use the correct placeholders for the configuration warnings

The same placeholder mismatch occurs at all three warning calls in loadMppDataExchangeMaxPayloadSize():

1. Non-positive configured value, lines 3102–3106.
2. Below the minimum value, lines 3109–3114.
3. Above the maximum value, lines 3119–3124.

All three new message constants contain %d, but LOGGER.warn(template, args...) performs SLF4J {} substitution. For example, the first warning is emitted as mpp_data_exchange_max_payload_size_in_bytes (%d) is not positive, using default value %d, without the configured or effective values. Both the English and Chinese templates are affected. Please use {} in the templates, or explicitly apply String.format() before logging, so operators can see which values were rejected and applied.