[fix](fe) Fix broken pipe risk on stream load redirect with unconsumed request body (#63332)

### What problem does this PR solve?

Issue Number: close #63325

Problem Summary:

**Problem**
- Starting from Doris `3.1.3`, FE uses `Jetty 12`, and this introduced a
compatibility change in the Stream Load redirect path.
- When a Stream Load request is sent to FE, FE may return `307 Temporary
Redirect` before the request body is fully consumed. Under `Jetty 12`,
this behavior is more likely to cause early connection close or reset
while the client is still writing the request body.
- As a result, some `HTTP/1.1` streaming clients may observe errors such
as `BrokenPipeError` or `ConnectionResetError` when sending Stream Load
requests through FE.
- The problem is more visible with chunked uploads, higher network
latency, and clients that continue sending request body data before
fully processing the redirect response.
- In short, this is a compatibility regression introduced by the `Jetty
12` upgrade in Doris `3.1.3` and later.

**Fix**
- We keep the existing FE-to-BE redirect architecture unchanged, so FE
still redirects Stream Load requests to BE instead of proxying the full
request body.
- We add a bounded request-body drain step on the FE Stream Load
redirect path:
  - FE first writes the `307 Temporary Redirect` response.
- FE then drains and discards only a bounded amount of the remaining
request body.
- This provides a small compatibility window for in-flight client writes
and reduces the chance of early connection reset.
- We also apply the same handling to token-authenticated Stream Load
requests, so both password-authenticated and token-authenticated paths
behave consistently.
- In addition, we expose Jetty's unconsumed request content read setting
through FE configuration and apply it to HTTP connectors, so operators
can tune Jetty behavior for redirect scenarios where the request body is
not fully consumed.
- To make the compatibility path effective out of the box, this PR also
enables the bounded drain path by default with a `1GB` drain limit and a
`1000ms` idle wait window.

**New Configurations**
- `jetty_server_max_unconsumed_request_content_reads`
- Controls how many extra reads Jetty performs for unconsumed request
content.
- `-1` means unlimited, `0` disables extra reads, and a positive value
sets the maximum number of read attempts.
  - Default value in this PR: `-1`.
- This helps tune Jetty behavior after the `Jetty 12` upgrade when FE
returns a response before the request body is fully consumed.

- `stream_load_redirect_bounded_drain_max_bytes`
- Controls the maximum number of request body bytes FE drains after
returning `307` for a Stream Load redirect.
  - `0` disables this compatibility logic.
- A positive value enables bounded draining and limits how much data FE
will discard.
  - Default value in this PR: `1GB`.

- `stream_load_redirect_bounded_drain_max_idle_time_ms`
- Controls how long FE waits for more readable request body data during
the bounded drain process.
  - `0` disables the extra idle wait.
- A positive value provides a small grace window for slow clients or
delayed body chunks, helping absorb in-flight writes without keeping the
connection open indefinitely.
  - Default value in this PR: `1000ms`.

**Test Result / Validation**
- Verified the behavior with the same Python `HTTP/1.1` chunked Stream
Load reproduction used during issue analysis.
- Reproduced requests were sent to FE with `Expect: 100-continue`,
`Transfer-Encoding: chunked`, and paced body streaming to maximize the
redirect race window.
- Baseline validation on Doris `3.0` (`9030` / FE `8030`):
  - `payload_mb=1`, `chunk_kb=1`, `sleep_ms=0`
  - `payload_mb=8`, `chunk_kb=16`, `sleep_ms=10`
  - Both requests returned normal `307 Temporary Redirect`.
- Validation on the fixed Doris `3.1.4` instance (`9034` / FE `8034`):
- Before enabling the bounded drain config, the same reproduction still
triggered `BrokenPipeError`.
  - After enabling the FE configs below:
    - `jetty_server_max_unconsumed_request_content_reads = -1`
    - `stream_load_redirect_bounded_drain_max_bytes = 16777216`
    - `stream_load_redirect_bounded_drain_max_idle_time_ms = 1000`
- The same two reproduction requests both returned normal `307 Temporary
Redirect`.
- No `BrokenPipeError` or `ConnectionResetError` was observed after the
config took effect.
- The PR now further updates the default bounded drain byte limit from
`16MB` to `1GB`, while keeping the default idle wait at `1000ms`, so the
compatibility path is enabled by default with a more generous drain
window.

**Performance Validation**
- I compared FE redirect response time between the Doris `3.0` baseline
instance (`9030`) and the fixed Doris `3.1.4` instance (`9034`).
- The goal was to check whether the additional bounded drain logic on FE
introduces a noticeable regression compared with the original Jetty 9
behavior.

**Test Setup**
- Reproduction tool: `tools/stream_load_redirect_repro.py`
- Target: FE endpoint on both instances
- Client mode: `httpclient`
- Common parameters:
  - `chunk_kb = 16`
  - `sleep_ms = 0`
- Payload sizes:
  - `32MB`
  - `128MB`
  - `512MB`
- Each case was executed `3` times on each instance, and the average
`elapsed_seconds` was used for comparison.

**Results**
- `32MB`
  - `9030`: `9.515s / 12.032s / 9.727s`
  - Average: `10.425s`
  - `9034`: `10.695s / 10.847s / 8.763s`
  - Average: `10.102s`
  - Difference: `9034` was `0.323s` faster, about `3.1%`

- `128MB`
  - `9030`: `37.174s / 34.111s / 37.090s`
  - Average: `36.125s`
  - `9034`: `38.910s / 36.423s / 38.337s`
  - Average: `37.890s`
  - Difference: `9034` was `1.765s` slower, about `4.9%`

- `512MB`
  - `9030`: `157.181s / 161.148s / 174.421s`
  - Average: `164.250s`
  - `9034`: `172.310s / 176.692s / 160.068s`
  - Average: `169.690s`
  - Difference: `9034` was `5.440s` slower, about `3.3%`

**Conclusion**
- Across all tested payload sizes, the difference between `9030` and
`9034` stayed within a small range, roughly `-3%` to `+5%`.
- Based on these measurements, the FE bounded drain logic does not show
a significant performance regression compared with the baseline FE
redirect behavior.
- In other words, the fix improves redirect compatibility while keeping
FE redirect response time at a similar level in normal request sizes.
---------

Co-authored-by: yaoxiao <yx136264032@163.com>

Author: wenzhenghu

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -380,6 +380,13 @@ public class Config extends ConfigBase {
     @ConfField(description = {"The maximum HTTP POST size of Jetty, in bytes, the default value is 100MB."})
     public static int jetty_server_max_http_post_size = 100 * 1024 * 1024;
 
+    @ConfField(description = {
+            "Jetty 在应用未消费完请求体时，额外尝试读取剩余内容的最大次数。"
+                    + "-1 表示不限制，0 表示不额外读取，正数表示最大读取次数。",
+            "The maximum number of extra reads Jetty performs for unconsumed request content. "
+                    + "-1 means unlimited, 0 means disabled, and a positive value limits the read attempts."})
+    public static int jetty_server_max_unconsumed_request_content_reads = -1;
+
     @ConfField(description = {"The maximum HTTP header size of Jetty, in bytes, the default value is 1MB."})
     public static int jetty_server_max_http_header_size = 1048576;
 
@@ -3328,6 +3335,23 @@ public static int metaServiceRpcRetryTimes() {
             + "public-private/public/private/direct/random-be and empty string."})
     public static String streamload_redirect_policy = "";
 
+    @ConfField(mutable = true, description = {
+            "Stream Load redirect 场景下，FE 在返回 307 后额外丢弃请求体的最大字节数。"
+                    + "0 表示关闭该兼容逻辑，正数表示最大丢弃字节数。",
+            "The maximum number of request body bytes FE drains after returning 307 for Stream Load redirects. "
+                    + "0 disables the compatibility logic, and a positive value sets the byte limit."})
+    // Enable a generous bounded drain window by default to preserve FE redirect compatibility on Jetty 12.
+    public static long stream_load_redirect_bounded_drain_max_bytes = 1024L * 1024 * 1024;
+
+    @ConfField(mutable = true, description = {
+            "Stream Load redirect 场景下，FE 在检测到请求体暂时无可读数据后继续等待的最大空闲时长，单位毫秒。"
+                    + "0 表示不额外等待，用于给慢客户端或分段到达的数据保留一个有限的缓冲窗口。",
+            "The maximum idle wait time in milliseconds after FE detects no readable request body bytes "
+                    + "during Stream Load redirect drain. 0 disables the extra idle wait, while a positive value "
+                    + "keeps a bounded grace window for slow clients or delayed request body chunks."})
+    // Keep a small grace period for delayed body chunks after FE has already written the redirect.
+    public static int stream_load_redirect_bounded_drain_max_idle_time_ms = 1000;
+
     @ConfField(mutable = true, description = {
             "Whether to enable group commit streamload BE forward feature in cloud mode. "
                     + "Solves the issue where LB random forwarding breaks group commit batching "

fe/fe-core/src/main/java/org/apache/doris/httpv2/config/WebServerFactoryCustomizerConfig.java

@@ -50,6 +50,9 @@ public void customize(ConfigurableJettyWebServerFactory factory) {
                     if (httpFactory != null) {
                         HttpConfiguration httpConfig = httpFactory.getHttpConfiguration();
                         httpConfig.setRequestHeaderSize(Config.jetty_server_max_http_header_size);
+                        // Apply the unconsumed request content read limit to every HTTP connector.
+                        httpConfig.setMaxUnconsumedRequestContentReads(
+                                Config.jetty_server_max_unconsumed_request_content_reads);
                     }
                 }
             }

fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/LoadAction.java

@@ -32,6 +32,7 @@
 import org.apache.doris.httpv2.entity.ResponseEntityBuilder;
 import org.apache.doris.httpv2.entity.RestBaseResult;
 import org.apache.doris.httpv2.exception.UnauthorizedException;
+import org.apache.doris.httpv2.util.StreamLoadRedirectDrainUtil;
 import org.apache.doris.load.StreamLoadHandler;
 import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.planner.GroupCommitPlanner;
@@ -59,8 +60,8 @@
 import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.servlet.view.RedirectView;
 
+import java.io.IOException;
 import java.net.InetAddress;
-import java.net.URI;
 import java.util.Enumeration;
 import java.util.List;
 import java.util.Optional;
@@ -131,7 +132,7 @@ public Object streamLoad(HttpServletRequest request,
             if (!checkClusterToken(authToken)) {
                 throw new UnauthorizedException("Invalid token: " + authToken);
             }
-            return executeWithClusterToken(request, db, table, true);
+            return executeWithClusterToken(request, response, db, table, true);
         } else {
             try {
                 executeCheckPassword(request, response);
@@ -190,8 +191,7 @@ public Object streamLoadWithSql(HttpServletRequest request, HttpServletResponse
             LOG.info("redirect load action to destination={}, label: {}",
                     redirectAddr.toString(), label);
 
-            RedirectView redirectView = redirectTo(request, redirectAddr);
-            return redirectView;
+            return createRedirectResponse(request, response, redirectAddr, true, null, null, label);
         } catch (Exception e) {
             return new RestBaseResult(e.getMessage());
         }
@@ -311,11 +311,17 @@ private Object executeWithoutPassword(HttpServletRequest request,
                         redirectAddr.toString(), isStreamLoad, dbName, tableName, label);
             }
 
-            RedirectView redirectView = redirectTo(request, redirectAddr);
-            return redirectView;
+            return createRedirectResponse(request, response, redirectAddr, isStreamLoad, dbName, tableName, label);
         } catch (StreamLoadForwardException e) {
-            // Special handling for stream load forwarding
-            return e.getRedirectView();
+            // Handle IOException from redirect response generation in the forwarding path.
+            try {
+                return createRedirectResponse(request, response, e.getRedirectView(),
+                        isStreamLoad, db, table, label);
+            } catch (IOException ioException) {
+                LOG.warn("stream load forward redirect failed, stream: {}, db: {}, tbl: {}, label: {}, err: {}",
+                        isStreamLoad, db, table, label, ioException.getMessage());
+                return new RestBaseResult(ioException.getMessage());
+            }
         } catch (Exception e) {
             LOG.warn("load failed, stream: {}, db: {}, tbl: {}, label: {}, err: {}",
                     isStreamLoad, db, table, label, e.getMessage());
@@ -592,7 +598,7 @@ private Pair<String, Integer> splitHostAndPort(String hostPort) throws AnalysisE
     // AuditlogPlugin should be re-disigned carefully, and blow method focuses on
     // temporarily addressing the users' needs for audit logs.
     // So this function is not widely tested under general scenario
-    private Object executeWithClusterToken(HttpServletRequest request, String db,
+    private Object executeWithClusterToken(HttpServletRequest request, HttpServletResponse response, String db,
             String table, boolean isStreamLoad) {
         try {
             ConnectContext ctx = new ConnectContext();
@@ -635,29 +641,7 @@ private Object executeWithClusterToken(HttpServletRequest request, String db,
                             + "stream: {}, db: {}, tbl: {}, label: {}",
                     redirectAddr.toString(), isStreamLoad, dbName, tableName, label);
 
-            URI urlObj = null;
-            URI resultUriObj = null;
-            String urlStr = request.getRequestURI();
-            String userInfo = null;
-
-            try {
-                urlObj = new URI(urlStr);
-                resultUriObj = new URI("http", userInfo, redirectAddr.getHostname(),
-                        redirectAddr.getPort(), urlObj.getPath(), "", null);
-            } catch (Exception e) {
-                throw new RuntimeException(e);
-            }
-            String redirectUrl = resultUriObj.toASCIIString();
-            if (!Strings.isNullOrEmpty(request.getQueryString())) {
-                redirectUrl += request.getQueryString();
-            }
-            LOG.info("Redirect url: {}", "http://" + redirectAddr.getHostname() + ":"
-                    + redirectAddr.getPort() + urlObj.getPath());
-            RedirectView redirectView = new RedirectView(redirectUrl);
-            redirectView.setContentType("text/html;charset=utf-8");
-            redirectView.setStatusCode(org.springframework.http.HttpStatus.TEMPORARY_REDIRECT);
-
-            return redirectView;
+            return createRedirectResponse(request, response, redirectAddr, isStreamLoad, dbName, tableName, label);
         } catch (Exception e) {
             LOG.warn("Failed to execute stream load with cluster token, {}", e.getMessage(), e);
             return new RestBaseResult(e.getMessage());
@@ -677,6 +661,80 @@ private String getAllHeaders(HttpServletRequest request) {
         return headers.toString();
     }
 
+    private Object createRedirectResponse(HttpServletRequest request, HttpServletResponse response,
+            TNetworkAddress redirectAddr, boolean isStreamLoad, String dbName, String tableName, String label)
+            throws IOException {
+        String redirectUrl = buildRedirectUrl(request, redirectAddr);
+        if (!shouldUseBoundedDrainForStreamLoad(isStreamLoad)) {
+            return redirectTo(request, redirectAddr);
+        }
+        writeTemporaryRedirect(response, redirectUrl);
+        DrainDecision drainDecision = decideDrainDecisionForStreamLoadRedirect(request);
+        if (drainDecision != DrainDecision.DRAIN) {
+            LOG.info("skip bounded drain after stream load redirect, target: {}, db: {}, tbl: {}, label: {},"
+                            + " reason: {}",
+                    redirectAddr, dbName, tableName, label, drainDecision);
+            return null;
+        }
+        drainStreamLoadRequestBodyAfterRedirect(request, redirectAddr.toString(), dbName, tableName, label);
+        return null;
+    }
+
+    private Object createRedirectResponse(HttpServletRequest request, HttpServletResponse response,
+            RedirectView redirectView, boolean isStreamLoad, String dbName, String tableName, String label)
+            throws IOException {
+        if (!shouldUseBoundedDrainForStreamLoad(isStreamLoad)) {
+            return redirectView;
+        }
+        writeTemporaryRedirect(response, redirectView.getUrl());
+        DrainDecision drainDecision = decideDrainDecisionForStreamLoadRedirect(request);
+        if (drainDecision != DrainDecision.DRAIN) {
+            LOG.info("skip bounded drain after stream load redirect, target: {}, db: {}, tbl: {}, label: {},"
+                            + " reason: {}",
+                    redirectView.getUrl(), dbName, tableName, label, drainDecision);
+            return null;
+        }
+        drainStreamLoadRequestBodyAfterRedirect(request, redirectView.getUrl(), dbName, tableName, label);
+        return null;
+    }
+
+    private boolean shouldUseBoundedDrainForStreamLoad(boolean isStreamLoad) {
+        return isStreamLoad && Config.stream_load_redirect_bounded_drain_max_bytes > 0;
+    }
+
+    // Skip the bounded drain for header-only probes and oversized fixed-length bodies.
+    private DrainDecision decideDrainDecisionForStreamLoadRedirect(HttpServletRequest request) {
+        long contentLength = request.getContentLengthLong();
+        String transferEncoding = request.getHeader(HttpHeaderNames.TRANSFER_ENCODING.toString());
+        if (contentLength <= 0 && Strings.isNullOrEmpty(transferEncoding)) {
+            return DrainDecision.SKIP_NO_REQUEST_BODY;
+        }
+        if (contentLength > Config.stream_load_redirect_bounded_drain_max_bytes) {
+            return DrainDecision.SKIP_CONTENT_LENGTH_EXCEEDS_MAX_BYTES;
+        }
+        return DrainDecision.DRAIN;
+    }
+
+    private void drainStreamLoadRequestBodyAfterRedirect(HttpServletRequest request, String redirectTarget,
+            String dbName, String tableName, String label) {
+        long drainLimit = Config.stream_load_redirect_bounded_drain_max_bytes;
+        LOG.info("write stream load redirect and start bounded drain, target: {}, db: {}, tbl: {}, label: {},"
+                        + " max_drain_bytes: {}",
+                redirectTarget, dbName, tableName, label, drainLimit);
+        StreamLoadRedirectDrainUtil.DrainResult drainResult =
+                StreamLoadRedirectDrainUtil.drainRequestBodyAfterRedirect(request, drainLimit);
+        LOG.info("finish bounded drain after stream load redirect, target: {}, db: {}, tbl: {}, label: {},"
+                        + " drained_bytes: {}, elapsed_ms: {}, exit_reason: {}",
+                redirectTarget, dbName, tableName, label, drainResult.getDrainedBytes(),
+                drainResult.getElapsedMillis(), drainResult.getExitReason());
+    }
+
+    private enum DrainDecision {
+        SKIP_NO_REQUEST_BODY,
+        SKIP_CONTENT_LENGTH_EXCEEDS_MAX_BYTES,
+        DRAIN
+    }
+
     private boolean isSensitiveHeader(String headerName) {
         return "Authorization".equalsIgnoreCase(headerName)
                 || "Proxy-Authorization".equalsIgnoreCase(headerName)
@@ -738,34 +796,14 @@ private Backend selectBackendForGroupCommit(String clusterName, HttpServletReque
      */
     private RedirectView redirectToStreamLoadForward(HttpServletRequest request, TNetworkAddress addr,
             String forwardTarget) {
-        URI urlObj = null;
-        URI resultUriObj = null;
-        String urlStr = request.getRequestURI();
-        String userInfo = null;
-        String modifiedPath = null;
-
-        if (!Strings.isNullOrEmpty(request.getHeader("Authorization"))) {
-            ActionAuthorizationInfo authInfo = getAuthorizationInfo(request);
-            userInfo = authInfo.fullUserName + ":" + authInfo.password;
-        }
-        try {
-            urlObj = new URI(urlStr);
-            // Replace _stream_load with _stream_load_forward in the path
-            modifiedPath = urlObj.getPath().replace("/_stream_load", "/_stream_load_forward");
-            resultUriObj = new URI("http", userInfo, addr.getHostname(),
-                    addr.getPort(), modifiedPath, "", null);
-        } catch (Exception e) {
-            throw new RuntimeException(e);
-        }
-        String redirectUrl = resultUriObj.toASCIIString();
-
-        // Add forward_to parameter (note: toASCIIString() already includes '?' due to empty query)
+        // Replace _stream_load with _stream_load_forward in the path.
+        String modifiedPath = request.getRequestURI().replace("/_stream_load", "/_stream_load_forward");
         String queryString = request.getQueryString();
+        String redirectQuery = "forward_to=" + forwardTarget;
         if (!Strings.isNullOrEmpty(queryString)) {
-            redirectUrl += queryString + "&forward_to=" + forwardTarget;
-        } else {
-            redirectUrl += "forward_to=" + forwardTarget;
+            redirectQuery = queryString + "&" + redirectQuery;
         }
+        String redirectUrl = buildRedirectUrl(request, addr, modifiedPath, redirectQuery);
 
         LOG.info("Redirect stream load forward url: {}, forward_to: {}",
                 "http://" + addr.getHostname() + ":" + addr.getPort() + modifiedPath, forwardTarget);

fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/RestBaseController.java

@@ -88,36 +88,49 @@ public ActionAuthorizationInfo executeCheckPassword(HttpServletRequest request,
         return authInfo;
     }
 
-    public RedirectView redirectTo(HttpServletRequest request, TNetworkAddress addr) {
-        RedirectView redirectView = new RedirectView(getRedirectUrL(request, addr));
-        redirectView.setContentType("text/html;charset=utf-8");
-        redirectView.setStatusCode(org.springframework.http.HttpStatus.TEMPORARY_REDIRECT);
-        return redirectView;
+    protected String buildRedirectUrl(HttpServletRequest request, TNetworkAddress addr) {
+        return buildRedirectUrl(request, addr, request.getRequestURI(), request.getQueryString());
     }
 
-    public String getRedirectUrL(HttpServletRequest request, TNetworkAddress addr) {
-        URI urlObj = null;
-        URI resultUriObj = null;
-        String urlStr = request.getRequestURI();
+    protected String buildRedirectUrl(HttpServletRequest request, TNetworkAddress addr, String requestPath,
+            String queryString) {
         String userInfo = null;
         if (!Strings.isNullOrEmpty(request.getHeader("Authorization"))) {
             ActionAuthorizationInfo authInfo = getAuthorizationInfo(request);
             userInfo = authInfo.fullUserName + ":" + authInfo.password;
         }
         try {
-            urlObj = new URI(urlStr);
-            resultUriObj = new URI(request.getScheme(), userInfo, addr.getHostname(),
-                    addr.getPort(), urlObj.getPath(), "", null);
+            // Preserve the original request path to avoid re-encoding an already encoded URI path.
+            URI authorityUri = new URI(request.getScheme(), userInfo, addr.getHostname(),
+                    addr.getPort(), null, null, null);
+            String redirectUrl = authorityUri.toASCIIString() + requestPath;
+            if (!Strings.isNullOrEmpty(queryString)) {
+                redirectUrl += "?" + queryString;
+            }
+            LOG.info("Redirect url: {}", request.getScheme() + "://" + addr.getHostname() + ":"
+                    + addr.getPort() + requestPath);
+            return redirectUrl;
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
-        String redirectUrl = resultUriObj.toASCIIString();
-        if (!Strings.isNullOrEmpty(request.getQueryString())) {
-            redirectUrl += request.getQueryString();
-        }
-        LOG.info("Redirect url: {}", request.getScheme() + "://" + addr.getHostname() + ":"
-                + addr.getPort() + urlObj.getPath());
-        return redirectUrl;
+    }
+
+    protected void writeTemporaryRedirect(HttpServletResponse response, String redirectUrl) throws IOException {
+        response.setContentType("text/html;charset=utf-8");
+        response.setStatus(HttpStatus.TEMPORARY_REDIRECT.value());
+        response.setHeader("Location", redirectUrl);
+        response.flushBuffer();
+    }
+
+    public RedirectView redirectTo(HttpServletRequest request, TNetworkAddress addr) {
+        RedirectView redirectView = new RedirectView(buildRedirectUrl(request, addr));
+        redirectView.setContentType("text/html;charset=utf-8");
+        redirectView.setStatusCode(org.springframework.http.HttpStatus.TEMPORARY_REDIRECT);
+        return redirectView;
+    }
+
+    public String getRedirectUrL(HttpServletRequest request, TNetworkAddress addr) {
+        return buildRedirectUrl(request, addr);
     }
 
     public RedirectView redirectToObj(String sign) throws URISyntaxException {