You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "wenzhenghu (via GitHub)" <gi...@apache.org> on 2026/05/17 13:18:36 UTC
[I] Stream Load via FE may fail with broken pipe on master while Doris 3.0 works [doris]
wenzhenghu opened a new issue, #63325:
URL: https://github.com/apache/doris/issues/63325
## Problem Description
When using Python low-level HTTP clients to perform Stream Load through FE, Doris `master` may fail with connection errors such as:
- `BrokenPipeError(32, 'Broken pipe')`
- `ConnectionResetError(54, 'Connection reset by peer')`
The same client behavior against Doris `3.0` FE can still receive a normal `307 Temporary Redirect` response without triggering the connection error.
This indicates a compatibility regression in the FE Stream Load redirect path compared with Doris `3.0`.
## Impact
This problem mainly affects clients that:
- send Stream Load to FE instead of BE
- use `HTTP/1.1`
- start sending request body before fully processing the FE redirect
- use chunked transfer or generator-based streaming
- run in higher RTT environments
Typical affected clients include:
- Python `http.client`
- Python `requests` in some streaming modes
- Logstash HTTP-based output plugins
## Reproduction
### Target Instances
Two Doris instances were used for comparison on the same host:
#### Doris 3.0
- MySQL: `172.16.0.90:9030`
- FE HTTP: `172.16.0.90:8030`
- BE HTTP: `172.16.0.90:8040`
- Version: `doris-3.0.8-rc01-53c80683e85`
#### Doris master
- MySQL: `172.16.0.90:9034`
- FE HTTP: `172.16.0.90:8034`
- BE HTTP: `172.16.0.90:8044`
- Version: `doris-0.0.0-b06684a15d5`
### Reproduction Script
A Python reproduction script was prepared using `http.client` and chunked upload to simulate a client that keeps sending request body while FE responds with redirect:
- client: Python `http.client`
- request type: `PUT /api/{db}/{table}/_stream_load`
- transfer mode: `Transfer-Encoding: chunked`
- header: `Expect: 100-continue`
### Reproduction Result on Doris 3.0
Request to FE:
```text
http://172.16.0.90:8030/api/wzh/stream_load_redirect_repro/_stream_load
```
Result:
```json
{
"target": "fe",
"url": "http://172.16.0.90:8030/api/wzh/stream_load_redirect_repro/_stream_load",
"client": "httpclient",
"status_code": 307,
"elapsed_seconds": 18.205,
"headers": {
"Location": "http://root:@172.16.0.90:8040/api/wzh/stream_load_redirect_repro/_stream_load?",
"Connection": "close"
},
"body": ""
}
```
### Reproduction Result on Doris master
Request to FE:
```text
http://172.16.0.90:8034/api/wzhtest/stream_load_redirect_repro/_stream_load
```
Result:
```json
{
"target": "fe",
"url": "http://172.16.0.90:8034/api/wzhtest/stream_load_redirect_repro/_stream_load",
"client": "httpclient",
"elapsed_seconds": 0.605,
"exception_type": "ConnectionResetError",
"exception": "ConnectionResetError(54, 'Connection reset by peer')"
}
```
In another run with a larger payload and paced chunk sending, Doris master also reproduced:
```json
{
"exception_type": "BrokenPipeError",
"exception": "BrokenPipeError(32, 'Broken pipe')"
}
```
## Comparison Between Doris 3.0 and master
Using the same host, same network, same Python client style, and same reproduction approach:
- Doris `3.0` FE returns normal `307 redirect`
- Doris `master` FE closes/resets the connection early
- This strongly suggests a compatibility regression in the FE Stream Load redirect path
## Root Cause Analysis
The current FE Stream Load path on `master` still behaves as:
1. validate request
2. select target BE
3. immediately return `307 Temporary Redirect`
4. do not consume the request body
At the same time, FE is now running on:
- `Spring Boot 3`
- `Spring Framework 6`
- `Jetty 12`
Jetty 12 is more sensitive/strict in the `HTTP/1.1` case where the application returns a response before consuming the request body.
If the client is still sending request body data when FE already redirects/closes the connection, the client may observe:
- `broken pipe`
- `connection reset by peer`
This matches the observed difference between Doris `3.0` and `master`.
## Proposed Fix
Two service-side improvements are proposed:
### 1. Expose Jetty unconsumed request reads as FE config
Expose Jetty's:
- `maxUnconsumedRequestContentReads`
as an FE configuration, for example:
- `jetty_server_max_unconsumed_request_content_reads`
and apply it to FE `HttpConfiguration`.
### 2. Add bounded drain compatibility logic for Stream Load redirect
On FE Stream Load redirect path:
- keep FE -> BE redirect architecture unchanged
- do not proxy the full request body through FE
- after writing `307`, drain/discard a bounded amount of remaining request body
- control it through an FE config, for example:
- `stream_load_redirect_bounded_drain_max_bytes`
This should improve compatibility with older/general-purpose HTTP clients while preserving the current FE/BE architecture.
## Environment
- macOS client
- Python 3.14
- PyMySQL + Python `http.client`
- same host and same network path for both Doris instances
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [I] [Bug] Stream Load via FE may fail with broken pipe on master while Doris 3.0 works [doris]
Posted by "wenzhenghu (via GitHub)" <gi...@apache.org>.
wenzhenghu commented on issue #63325:
URL: https://github.com/apache/doris/issues/63325#issuecomment-4470827857
Adding the reproduction script and exact command lines here for completeness.
The real host/IP is intentionally anonymized as `<REDACTED_HOST>`.
## Reproduction Script
```python
#!/usr/bin/env python3
import argparse
import base64
import http.client
import json
import sys
import time
import uuid
import pymysql
import requests
def parse_args():
parser = argparse.ArgumentParser(
description="Send a chunked stream load request to FE or BE with controllable pacing."
)
parser.add_argument("--host", default="<REDACTED_HOST>", help="Doris FE/BE host")
parser.add_argument("--mysql-port", type=int, default=9034, help="Doris MySQL port")
parser.add_argument("--fe-http-port", type=int, default=8034, help="Doris FE HTTP port")
parser.add_argument("--be-http-port", type=int, default=8044, help="Doris BE HTTP port")
parser.add_argument("--user", default="root", help="Doris user")
parser.add_argument("--password", default="", help="Doris password")
parser.add_argument("--db", default="wzhtest", help="Target database")
parser.add_argument("--table", default="stream_load_redirect_repro", help="Target table")
parser.add_argument(
"--target",
choices=["fe", "be"],
default="fe",
help="Choose whether the request goes to FE or BE HTTP port",
)
parser.add_argument(
"--payload-mb",
type=int,
default=8,
help="Approximate payload size in MiB generated by the script",
)
parser.add_argument(
"--chunk-kb",
type=int,
default=16,
help="Application chunk size in KiB for generator-based upload",
)
parser.add_argument(
"--sleep-ms",
type=int,
default=0,
help="Sleep duration between chunks to enlarge the redirect race window",
)
parser.add_argument(
"--connect-timeout",
type=int,
default=5,
help="HTTP connect timeout in seconds",
)
parser.add_argument(
"--read-timeout",
type=int,
default=600,
help="HTTP read timeout in seconds",
)
parser.add_argument(
"--label-prefix",
default="stream_load_redirect_repro",
help="Prefix used to build a unique stream load label",
)
parser.add_argument(
"--client",
choices=["requests", "httpclient"],
default="requests",
help="Choose the Python HTTP stack used to send the stream load request",
)
parser.add_argument(
"--truncate-before",
action="store_true",
help="Truncate the target table before sending the stream load request",
)
parser.add_argument(
"--show-row-count",
action="store_true",
help="Query and print the target table row count after the request",
)
return parser.parse_args()
def build_auth_header(user, password):
token = base64.b64encode(f"{user}:{password}".encode("utf-8")).decode("ascii")
return f"Basic {token}"
def connect_mysql(args):
return pymysql.connect(
host=args.host,
port=args.mysql_port,
user=args.user,
password=args.password,
connect_timeout=5,
read_timeout=10,
write_timeout=10,
autocommit=True,
)
def truncate_table(args):
with connect_mysql(args) as conn:
with conn.cursor() as cur:
cur.execute(f"TRUNCATE TABLE `{args.db}`.`{args.table}`")
def query_row_count(args):
with connect_mysql(args) as conn:
with conn.cursor() as cur:
cur.execute(f"SELECT COUNT(*) FROM `{args.db}`.`{args.table}`")
return cur.fetchone()[0]
def build_csv_chunk(chunk_bytes, chunk_index):
rows = []
current = 0
row_index = chunk_index * 100000
payload_width = max(8, min(512, chunk_bytes // 8))
while True:
row = f"{row_index},payload_{chunk_index}_{'x' * payload_width}\n"
row_size = len(row.encode("utf-8"))
if rows and current + row_size > chunk_bytes:
break
rows.append(row)
current += row_size
row_index += 1
if current >= chunk_bytes:
break
return "".join(rows).encode("utf-8")
def chunked_csv_generator(total_bytes, chunk_bytes, sleep_ms):
sent = 0
chunk_index = 0
while sent < total_bytes:
remaining = total_bytes - sent
current_size = min(chunk_bytes, remaining)
yield build_csv_chunk(current_size, chunk_index)
sent += current_size
chunk_index += 1
if sleep_ms > 0:
time.sleep(sleep_ms / 1000.0)
def stream_load(args):
port = args.fe_http_port if args.target == "fe" else args.be_http_port
url = f"http://{args.host}:{port}/api/{args.db}/{args.table}/_stream_load"
label = f"{args.label_prefix}_{args.target}_{uuid.uuid4().hex[:12]}"
total_bytes = args.payload_mb * 1024 * 1024
chunk_bytes = args.chunk_kb * 1024
headers = {
"Authorization": build_auth_header(args.user, args.password),
"Expect": "100-continue",
"column_separator": ",",
"label": label,
}
if args.client == "httpclient":
return stream_load_with_httpclient(args, url, label, total_bytes, chunk_bytes, headers)
session = requests.Session()
started = time.time()
try:
response = session.put(
url=url,
headers=headers,
data=chunked_csv_generator(total_bytes, chunk_bytes, args.sleep_ms),
allow_redirects=False,
timeout=(args.connect_timeout, args.read_timeout),
)
elapsed = time.time() - started
print(json.dumps(
{
"target": args.target,
"url": url,
"label": label,
"status_code": response.status_code,
"elapsed_seconds": round(elapsed, 3),
"headers": dict(response.headers),
"body": response.text[:2000],
},
ensure_ascii=False,
indent=2,
))
return 0
except Exception as exc:
elapsed = time.time() - started
print(json.dumps(
{
"target": args.target,
"url": url,
"label": label,
"elapsed_seconds": round(elapsed, 3),
"exception_type": type(exc).__name__,
"exception": repr(exc),
},
ensure_ascii=False,
indent=2,
))
return 1
finally:
session.close()
def stream_load_with_httpclient(args, url, label, total_bytes, chunk_bytes, headers):
port = args.fe_http_port if args.target == "fe" else args.be_http_port
parsed_path = f"/api/{args.db}/{args.table}/_stream_load"
conn = http.client.HTTPConnection(args.host, port, timeout=args.read_timeout)
started = time.time()
try:
conn.putrequest("PUT", parsed_path)
conn.putheader("Authorization", headers["Authorization"])
conn.putheader("Expect", headers["Expect"])
conn.putheader("column_separator", headers["column_separator"])
conn.putheader("label", label)
conn.putheader("Transfer-Encoding", "chunked")
conn.endheaders()
for chunk in chunked_csv_generator(total_bytes, chunk_bytes, args.sleep_ms):
conn.send(f"{len(chunk):X}\r\n".encode("ascii"))
conn.send(chunk)
conn.send(b"\r\n")
conn.send(b"0\r\n\r\n")
response = conn.getresponse()
body = response.read().decode("utf-8", errors="replace")
elapsed = time.time() - started
print(json.dumps(
{
"target": args.target,
"url": url,
"label": label,
"client": args.client,
"status_code": response.status,
"elapsed_seconds": round(elapsed, 3),
"headers": dict(response.getheaders()),
"body": body[:2000],
},
ensure_ascii=False,
indent=2,
))
return 0
except Exception as exc:
elapsed = time.time() - started
print(json.dumps(
{
"target": args.target,
"url": url,
"label": label,
"client": args.client,
"elapsed_seconds": round(elapsed, 3),
"exception_type": type(exc).__name__,
"exception": repr(exc),
},
ensure_ascii=False,
indent=2,
))
return 1
finally:
conn.close()
def main():
args = parse_args()
if args.truncate_before:
truncate_table(args)
rc = stream_load(args)
if args.show_row_count:
try:
row_count = query_row_count(args)
print(json.dumps({"row_count": row_count}, ensure_ascii=False))
except Exception as exc:
print(json.dumps({"row_count_error": repr(exc)}, ensure_ascii=False))
return rc
if __name__ == "__main__":
sys.exit(main())
```
## Reproduction Commands
### Doris 3.0 FE: receives a normal 307 redirect
```bash
python stream_load_redirect_repro.py \
--client httpclient \
--host <REDACTED_HOST> \
--mysql-port 9030 \
--fe-http-port 8030 \
--be-http-port 8040 \
--db wzh \
--table stream_load_redirect_repro \
--target fe \
--payload-mb 1 \
--chunk-kb 1 \
--sleep-ms 0 \
--truncate-before \
--show-row-count
```
Expected result: FE returns `307 Temporary Redirect`.
### Doris master FE: reproduces connection reset / broken pipe
```bash
python stream_load_redirect_repro.py \
--client httpclient \
--host <REDACTED_HOST> \
--mysql-port 9034 \
--fe-http-port 8034 \
--be-http-port 8044 \
--db wzhtest \
--table stream_load_redirect_repro \
--target fe \
--payload-mb 1 \
--chunk-kb 1 \
--sleep-ms 0 \
--truncate-before \
--show-row-count
```
Typical result on `master`: `ConnectionResetError`.
The following variant can reproduce `BrokenPipeError` more aggressively by increasing the write window:
```bash
python stream_load_redirect_repro.py \
--client httpclient \
--host <REDACTED_HOST> \
--mysql-port 9034 \
--fe-http-port 8034 \
--be-http-port 8044 \
--db wzhtest \
--table stream_load_redirect_repro \
--target fe \
--payload-mb 8 \
--chunk-kb 16 \
--sleep-ms 10 \
--truncate-before \
--show-row-count
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [I] [Bug] Stream Load via FE may fail with broken pipe on master while Doris 3.0 works [doris]
Posted by "liaoxin01 (via GitHub)" <gi...@apache.org>.
liaoxin01 closed issue #63325: [Bug] Stream Load via FE may fail with broken pipe on master while Doris 3.0 works
URL: https://github.com/apache/doris/issues/63325
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org