This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new e6c097391d [Fix][Connector-V2] Fix Doris stream load waitForContinue
timeout when FE redirect is slow (#11330)
e6c097391d is described below
commit e6c097391d03c5122ad950c6da25a9263267e7bc
Author: gnehil <[email protected]>
AuthorDate: Sat Jul 25 20:18:11 2026 +0800
[Fix][Connector-V2] Fix Doris stream load waitForContinue timeout when FE
redirect is slow (#11330)
---
.../doris/sink/writer/DorisStreamLoad.java | 14 +-
.../seatunnel/connectors/doris/util/HttpUtil.java | 43 ++++-
.../connectors/doris/util/HttpUtilTest.java | 175 +++++++++++++++++++++
3 files changed, 224 insertions(+), 8 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisStreamLoad.java
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisStreamLoad.java
index ca223a0b35..01e7559d4c 100644
---
a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisStreamLoad.java
+++
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisStreamLoad.java
@@ -85,6 +85,10 @@ public class DorisStreamLoad implements Serializable {
private final RecordStream recordStream;
@Getter private Future<CloseableHttpResponse> pendingLoadFuture;
private final CloseableHttpClient httpClient;
+ // Dedicated client for the stream-load upload path. Unlike the control
requests that share
+ // httpClient, the upload sends a non-repeatable body and needs HttpUtil's
longer
+ // waitForContinue window so a slow FE 307 redirect does not force the
body to FE first.
+ private final CloseableHttpClient streamLoadHttpClient;
private final ExecutorService executorService;
private volatile boolean loadBatchFirstRecord;
private volatile boolean loading = false;
@@ -113,6 +117,7 @@ public class DorisStreamLoad implements Serializable {
this.streamLoadProp = dorisSinkConfig.getStreamLoadProps();
this.enableDelete = dorisSinkConfig.getEnableDelete();
this.httpClient = httpClient;
+ this.streamLoadHttpClient = new HttpUtil().getStreamLoadHttpClient();
this.executorService =
new ThreadPoolExecutor(
1,
@@ -297,7 +302,7 @@ public class DorisStreamLoad implements Serializable {
() -> {
log.info("start execute load");
return HttpUtil.executeWithRedirectTracking(
- httpClient,
+ streamLoadHttpClient,
putBuilder.build(),
loadUrlStr,
directToBe,
@@ -356,6 +361,13 @@ public class DorisStreamLoad implements Serializable {
}
public void close() throws IOException {
+ if (null != streamLoadHttpClient) {
+ try {
+ streamLoadHttpClient.close();
+ } catch (IOException e) {
+ throw new IOException("Closing streamLoadHttpClient failed.",
e);
+ }
+ }
if (null != httpClient) {
try {
httpClient.close();
diff --git
a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/util/HttpUtil.java
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/util/HttpUtil.java
index 9a462e75eb..8917fc37dc 100644
---
a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/util/HttpUtil.java
+++
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/util/HttpUtil.java
@@ -27,6 +27,7 @@ import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.DefaultRedirectStrategy;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.impl.client.HttpClients;
+import org.apache.http.protocol.HttpRequestExecutor;
import org.apache.http.protocol.RequestContent;
import java.io.IOException;
@@ -35,21 +36,49 @@ import java.util.List;
/** util to build http client. */
public class HttpUtil {
+ // Stream load upload sends "Expect: 100-continue" and relies on FE's 307
redirect to reach
+ // BE. The default waitForContinue timeout of HttpRequestExecutor is only
3s: when FE is busy
+ // (heavy load / FullGC), the client stops waiting, sends the
non-repeatable request body to
+ // FE, and can no longer follow the late 307 redirect. Wait 60s for upload
traffic instead,
+ // aligned with doris-flink-connector. Only the stream-load upload client
needs this - the
+ // empty-entity control requests (commit / abort / pre-commit) keep the
default, so a slow FE
+ // does not stretch checkpoint completion / transaction cleanup.
+ private static final int WAIT_FOR_CONTINUE_TIMEOUT_MS = 60 * 1000;
+
+ private static final DefaultRedirectStrategy REDIRECT_STRATEGY =
+ new DefaultRedirectStrategy() {
+ @Override
+ protected boolean isRedirectable(String method) {
+ return true;
+ }
+ };
+
private final HttpClientBuilder httpClientBuilder =
HttpClients.custom()
- .setRedirectStrategy(
- new DefaultRedirectStrategy() {
- @Override
- protected boolean isRedirectable(String
method) {
- return true;
- }
- })
+ .setRedirectStrategy(REDIRECT_STRATEGY)
+ .addInterceptorLast(new RequestContent(true));
+
+ // Separate builder for the stream-load upload path, which sends a
non-repeatable body and
+ // needs a longer waitForContinue window than the default 3s.
+ private final HttpClientBuilder streamLoadHttpClientBuilder =
+ HttpClients.custom()
+ .setRequestExecutor(new
HttpRequestExecutor(WAIT_FOR_CONTINUE_TIMEOUT_MS))
+ .setRedirectStrategy(REDIRECT_STRATEGY)
.addInterceptorLast(new RequestContent(true));
public CloseableHttpClient getHttpClient() {
return httpClientBuilder.build();
}
+ /**
+ * Client for the stream-load upload path. Raises the waitForContinue
timeout so a slow FE 307
+ * redirect does not force the non-repeatable request body to be sent to
FE before the redirect
+ * arrives.
+ */
+ public CloseableHttpClient getStreamLoadHttpClient() {
+ return streamLoadHttpClientBuilder.build();
+ }
+
public static CloseableHttpResponse executeWithRedirectTracking(
CloseableHttpClient httpClient,
HttpUriRequest request,
diff --git
a/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/util/HttpUtilTest.java
b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/util/HttpUtilTest.java
new file mode 100644
index 0000000000..4c9eda266c
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/util/HttpUtilTest.java
@@ -0,0 +1,175 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.doris.util;
+
+import org.apache.seatunnel.connectors.doris.sink.HttpPutBuilder;
+
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.entity.InputStreamEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.util.EntityUtils;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import com.sun.net.httpserver.HttpServer;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class HttpUtilTest {
+
+ private static final String LOAD_PATH =
"/api/test_db/test_table/_stream_load";
+
+ /**
+ * A busy FE may return the 307 redirect later than the default 3s
waitForContinue timeout of
+ * HttpRequestExecutor. The stream load client must keep waiting for the
redirect instead of
+ * sending the non-repeatable request body to FE: once the body is
consumed, RedirectExec
+ * refuses to follow the redirect and returns the raw 307 response to the
caller.
+ */
+ @Test
+ @Timeout(30)
+ void testStreamLoadClientWaitsForSlowFeRedirect() throws Exception {
+ String payload = "{\"k1\":1}";
+ AtomicReference<String> bodyReceivedByBe = new AtomicReference<>();
+ HttpServer beServer = createBeServer(bodyReceivedByBe);
+ String location =
+ String.format("http://127.0.0.1:%s%s";,
beServer.getAddress().getPort(), LOAD_PATH);
+ try (SlowRedirectFeServer feServer = new SlowRedirectFeServer(4000,
location)) {
+ HttpPut put =
+ new HttpPutBuilder()
+ .setUrl(
+ String.format(
+ "http://127.0.0.1:%s%s";,
feServer.getPort(), LOAD_PATH))
+ .addCommonHeader()
+ .setEntity(
+ new InputStreamEntity(
+ new ByteArrayInputStream(
+
payload.getBytes(StandardCharsets.UTF_8))))
+ .build();
+ try (CloseableHttpClient httpClient = new
HttpUtil().getStreamLoadHttpClient();
+ CloseableHttpResponse response = httpClient.execute(put)) {
+ Assertions.assertEquals(200,
response.getStatusLine().getStatusCode());
+ EntityUtils.consume(response.getEntity());
+ }
+ Assertions.assertEquals(payload, bodyReceivedByBe.get());
+ } finally {
+ beServer.stop(0);
+ }
+ }
+
+ private HttpServer createBeServer(AtomicReference<String> bodyHolder)
throws IOException {
+ HttpServer server = HttpServer.create(new
InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext(
+ LOAD_PATH,
+ exchange -> {
+ bodyHolder.set(
+ new String(readAll(exchange.getRequestBody()),
StandardCharsets.UTF_8));
+ byte[] result =
"{\"Status\":\"Success\"}".getBytes(StandardCharsets.UTF_8);
+ exchange.sendResponseHeaders(200, result.length);
+ try (OutputStream outputStream =
exchange.getResponseBody()) {
+ outputStream.write(result);
+ }
+ });
+ server.setExecutor(Executors.newCachedThreadPool());
+ server.start();
+ return server;
+ }
+
+ private static byte[] readAll(InputStream inputStream) throws IOException {
+ ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+ byte[] chunk = new byte[1024];
+ int read;
+ while ((read = inputStream.read(chunk)) != -1) {
+ buffer.write(chunk, 0, read);
+ }
+ return buffer.toByteArray();
+ }
+
+ /**
+ * Raw-socket FE stub which reads the request headers and then delays the
307 redirect without
+ * sending "100 Continue". A com.sun HttpServer cannot simulate this: it
automatically replies
+ * "100 Continue" before the handler runs, so the client would never wait.
+ */
+ private static final class SlowRedirectFeServer implements AutoCloseable {
+ private final ServerSocket serverSocket;
+
+ SlowRedirectFeServer(long redirectDelayMs, String location) throws
IOException {
+ this.serverSocket = new ServerSocket(0, 1,
InetAddress.getByName("127.0.0.1"));
+ Thread thread =
+ new Thread(
+ () -> {
+ try (Socket socket = serverSocket.accept()) {
+
readRequestHeaders(socket.getInputStream());
+ Thread.sleep(redirectDelayMs);
+ String response =
+ "HTTP/1.1 307 Temporary
Redirect\r\n"
+ + "Location: "
+ + location
+ + "\r\n"
+ + "Content-Length: 0\r\n"
+ + "Connection: close\r\n"
+ + "\r\n";
+ OutputStream outputStream =
socket.getOutputStream();
+
outputStream.write(response.getBytes(StandardCharsets.UTF_8));
+ outputStream.flush();
+ } catch (Exception ignored) {
+ // client-side assertions fail the test
+ }
+ },
+ "slow-redirect-fe");
+ thread.setDaemon(true);
+ thread.start();
+ }
+
+ int getPort() {
+ return serverSocket.getLocalPort();
+ }
+
+ private static void readRequestHeaders(InputStream inputStream) throws
IOException {
+ int state = 0;
+ int b;
+ while (state < 4 && (b = inputStream.read()) != -1) {
+ if (b == '\r' && (state == 0 || state == 2)) {
+ state++;
+ } else if (b == '\n' && (state == 1 || state == 3)) {
+ state++;
+ } else {
+ state = 0;
+ }
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ serverSocket.close();
+ }
+ }
+}
