This is an automated email from the ASF dual-hosted git repository.
JNSimba pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-flink-connector.git
The following commit(s) were added to refs/heads/master by this push:
new 1050cd06 [Fix] Avoid repeated FE health checks during stream load
(#701)
1050cd06 is described below
commit 1050cd061ac66fda9c85ad875429295ba0c3de8f
Author: wudi <[email protected]>
AuthorDate: Tue Oct 13 10:49:46 2026 +0800
[Fix] Avoid repeated FE health checks during stream load (#701)
When automatic redirect is enabled without explicit BE nodes, the connector
uses FE nodes as stream load endpoints. Repeated health checks during backend
selection can block checkpoints when an FE is unreachable.
This change filters unavailable FE nodes once when BackendUtil is
initialized and keeps the existing round-robin selection and Flink recovery
behavior afterward. Explicit BE nodes and discovered BE nodes retain their
existing health-check behavior. Writer, committer, and batch stream load now
use the same BackendUtil creation path.
---
.../org/apache/doris/flink/sink/BackendUtil.java | 28 ++++++++--
.../flink/sink/batch/DorisBatchStreamLoad.java | 8 +--
.../doris/flink/sink/committer/DorisCommitter.java | 8 +--
.../apache/doris/flink/sink/TestBackendUtil.java | 60 ++++++++++++++++++++++
.../flink/sink/batch/TestDorisBatchStreamLoad.java | 3 ++
.../flink/sink/batch/TestDorisBatchWriter.java | 3 ++
.../flink/sink/committer/TestDorisCommitter.java | 3 ++
.../org/apache/doris/flink/sink/DorisSinkTest.java | 3 ++
.../org/apache/doris/flink/sink/DorisSinkTest.java | 3 ++
9 files changed, 102 insertions(+), 17 deletions(-)
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/BackendUtil.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/BackendUtil.java
index d728cb81..9c0dc7df 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/BackendUtil.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/BackendUtil.java
@@ -37,11 +37,13 @@ import java.net.URL;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.stream.Collectors;
public class BackendUtil {
private static final Logger LOG =
LoggerFactory.getLogger(BackendUtil.class);
private final List<BackendV2.BackendRowV2> backends;
private final DorisTlsOptions tlsOptions;
+ private final boolean isFe;
private long pos;
public BackendUtil(List<BackendV2.BackendRowV2> backends) {
@@ -49,8 +51,26 @@ public class BackendUtil {
}
public BackendUtil(List<BackendV2.BackendRowV2> backends, DorisTlsOptions
tlsOptions) {
- this.backends = backends;
+ this(backends, tlsOptions, false);
+ }
+
+ private BackendUtil(
+ List<BackendV2.BackendRowV2> backends, DorisTlsOptions tlsOptions,
boolean isFe) {
this.tlsOptions = tlsOptions;
+ this.isFe = isFe;
+ // Filter FE candidates once; runtime failures use the existing retry
or recovery path.
+ this.backends =
+ isFe
+ ? backends.stream()
+ .filter(
+ node ->
+ tryHttpConnection(
+
node.toBackendString(), tlsOptions))
+ .collect(Collectors.toList())
+ : backends;
+ if (isFe && this.backends.isEmpty()) {
+ throw new DorisRuntimeException("no available FE.");
+ }
this.pos = 0;
}
@@ -60,6 +80,7 @@ public class BackendUtil {
public BackendUtil(String beNodes, DorisTlsOptions tlsOptions) {
this.tlsOptions = tlsOptions;
+ this.isFe = false;
this.backends = initBackends(beNodes);
this.pos = 0;
}
@@ -90,7 +111,8 @@ public class BackendUtil {
} else {
return new BackendUtil(
RestService.getBackendsV2(dorisOptions, readOptions,
logger),
- dorisOptions.getTlsOptions());
+ dorisOptions.getTlsOptions(),
+ dorisOptions.isAutoRedirect());
}
}
@@ -105,7 +127,7 @@ public class BackendUtil {
backends.get((int) ((pos + subtaskId) % backends.size()));
pos++;
String res = backend.toBackendString();
- if (tryHttpConnection(res, tlsOptions)) {
+ if (isFe || tryHttpConnection(res, tlsOptions)) {
return res;
}
}
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchStreamLoad.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchStreamLoad.java
index 274e0b8c..90da87a6 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchStreamLoad.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchStreamLoad.java
@@ -29,7 +29,6 @@ import org.apache.doris.flink.cfg.DorisTlsOptions;
import org.apache.doris.flink.exception.DorisBatchLoadException;
import org.apache.doris.flink.exception.DorisRuntimeException;
import org.apache.doris.flink.rest.DorisUrlBuilder;
-import org.apache.doris.flink.rest.RestService;
import org.apache.doris.flink.rest.models.RespContent;
import org.apache.doris.flink.sink.BackendUtil;
import org.apache.doris.flink.sink.EscapeHandler;
@@ -125,12 +124,7 @@ public class DorisBatchStreamLoad implements Serializable {
LabelGenerator labelGenerator,
int subTaskId) {
this.tlsOptions = dorisOptions.getTlsOptions();
- this.backendUtil =
- StringUtils.isNotEmpty(dorisOptions.getBenodes())
- ? new BackendUtil(dorisOptions.getBenodes(),
tlsOptions)
- : new BackendUtil(
- RestService.getBackendsV2(dorisOptions,
dorisReadOptions, LOG),
- tlsOptions);
+ this.backendUtil = BackendUtil.getInstance(dorisOptions,
dorisReadOptions, LOG);
this.hostPort = backendUtil.getAvailableBackend();
this.username = dorisOptions.getUsername();
this.password = dorisOptions.getPassword();
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/committer/DorisCommitter.java
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/committer/DorisCommitter.java
index 4e3c6635..3da55443 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/committer/DorisCommitter.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/committer/DorisCommitter.java
@@ -27,7 +27,6 @@ import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.exception.DorisRuntimeException;
import org.apache.doris.flink.rest.DorisUrlBuilder;
-import org.apache.doris.flink.rest.RestService;
import org.apache.doris.flink.sink.BackendUtil;
import org.apache.doris.flink.sink.DorisCommittable;
import org.apache.doris.flink.sink.HttpPutBuilder;
@@ -88,12 +87,7 @@ public class DorisCommitter implements
Committer<DorisCommittable>, Closeable {
this.maxRetry = executionOptions.getMaxRetries();
this.ignoreCommitError = executionOptions.ignoreCommitError();
this.httpClient = client;
- this.backendUtil =
-
org.apache.commons.lang3.StringUtils.isNotEmpty(dorisOptions.getBenodes())
- ? new BackendUtil(dorisOptions.getBenodes(),
dorisOptions.getTlsOptions())
- : new BackendUtil(
- RestService.getBackendsV2(dorisOptions,
dorisReadOptions, LOG),
- dorisOptions.getTlsOptions());
+ this.backendUtil = BackendUtil.getInstance(dorisOptions,
dorisReadOptions, LOG);
}
@Override
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/sink/TestBackendUtil.java
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/sink/TestBackendUtil.java
index 15882881..d597e910 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/sink/TestBackendUtil.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/sink/TestBackendUtil.java
@@ -17,6 +17,8 @@
package org.apache.doris.flink.sink;
+import org.apache.doris.flink.cfg.DorisOptions;
+import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.cfg.DorisTlsOptions;
import org.apache.doris.flink.exception.DorisRuntimeException;
import org.apache.doris.flink.rest.models.BackendV2;
@@ -26,6 +28,7 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.mockito.MockedStatic;
+import org.slf4j.LoggerFactory;
import java.util.Arrays;
import java.util.List;
@@ -41,6 +44,63 @@ public class TestBackendUtil {
@Before
public void setUp() throws Exception {
backendUtilMockedStatic = mockStatic(BackendUtil.class);
+ backendUtilMockedStatic
+ .when(() -> BackendUtil.getInstance(any(), any(), any()))
+ .thenCallRealMethod();
+ }
+
+ @Test
+ public void testFeNodesAreFilteredOnlyDuringInitialization() throws
Exception {
+ DorisTlsOptions tlsOptions = tlsOptions();
+ DorisOptions options =
+ DorisOptions.builder()
+
.setFenodes("127.0.0.1:8030,127.0.0.2:8030,127.0.0.3:8030")
+ .setAutoRedirect(true)
+ .setTlsOptions(tlsOptions)
+ .build();
+ backendUtilMockedStatic
+ .when(() -> BackendUtil.tryHttpConnection("127.0.0.2:8030",
tlsOptions))
+ .thenReturn(true);
+ backendUtilMockedStatic
+ .when(() -> BackendUtil.tryHttpConnection("127.0.0.3:8030",
tlsOptions))
+ .thenReturn(true);
+
+ BackendUtil backendUtil =
+ BackendUtil.getInstance(
+ options, DorisReadOptions.defaults(),
LoggerFactory.getLogger(getClass()));
+ Assert.assertEquals(2, backendUtil.getBackends().size());
+
+ backendUtilMockedStatic
+ .when(() -> BackendUtil.tryHttpConnection(anyString(),
any(DorisTlsOptions.class)))
+ .thenReturn(false);
+ List<String> selected =
+ Arrays.asList(
+ backendUtil.getAvailableBackend(1),
backendUtil.getAvailableBackend(1));
+ Assert.assertTrue(selected.contains("127.0.0.2:8030"));
+ Assert.assertTrue(selected.contains("127.0.0.3:8030"));
+ Assert.assertEquals(selected.get(0),
backendUtil.getAvailableBackend(1));
+
+ // Reinitialization discovers the current availability instead of
reusing the old list.
+ backendUtilMockedStatic
+ .when(() -> BackendUtil.tryHttpConnection("127.0.0.3:8030",
tlsOptions))
+ .thenReturn(true);
+ BackendUtil recovered =
+ BackendUtil.getInstance(
+ options, DorisReadOptions.defaults(),
LoggerFactory.getLogger(getClass()));
+ Assert.assertEquals(1, recovered.getBackends().size());
+ Assert.assertEquals("127.0.0.3:8030", recovered.getAvailableBackend());
+ }
+
+ @Test(expected = DorisRuntimeException.class)
+ public void testNoAvailableFeFailsDuringInitialization() {
+ DorisOptions options =
+ DorisOptions.builder()
+ .setFenodes("127.0.0.1:8030,127.0.0.2:8030")
+ .setAutoRedirect(true)
+ .build();
+
+ BackendUtil.getInstance(
+ options, DorisReadOptions.defaults(),
LoggerFactory.getLogger(getClass()));
}
@Test
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/sink/batch/TestDorisBatchStreamLoad.java
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/sink/batch/TestDorisBatchStreamLoad.java
index d39268fe..31bef086 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/sink/batch/TestDorisBatchStreamLoad.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/sink/batch/TestDorisBatchStreamLoad.java
@@ -70,6 +70,9 @@ public class TestDorisBatchStreamLoad {
@Before
public void setUp() throws Exception {
backendUtilMockedStatic = mockStatic(BackendUtil.class);
+ backendUtilMockedStatic
+ .when(() -> BackendUtil.getInstance(any(), any(), any()))
+ .thenCallRealMethod();
backendUtilMockedStatic.when(() ->
BackendUtil.tryHttpConnection(any())).thenReturn(true);
backendUtilMockedStatic
.when(() -> BackendUtil.tryHttpConnection(any(), any()))
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/sink/batch/TestDorisBatchWriter.java
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/sink/batch/TestDorisBatchWriter.java
index 79340ce3..7ee479d4 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/sink/batch/TestDorisBatchWriter.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/sink/batch/TestDorisBatchWriter.java
@@ -42,6 +42,9 @@ public class TestDorisBatchWriter {
@Before
public void setUp() throws Exception {
backendUtilMockedStatic = mockStatic(BackendUtil.class);
+ backendUtilMockedStatic
+ .when(() -> BackendUtil.getInstance(any(), any(), any()))
+ .thenCallRealMethod();
backendUtilMockedStatic.when(() ->
BackendUtil.tryHttpConnection(any())).thenReturn(true);
backendUtilMockedStatic
.when(() -> BackendUtil.tryHttpConnection(any(), any()))
diff --git
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/sink/committer/TestDorisCommitter.java
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/sink/committer/TestDorisCommitter.java
index d1470c21..6350f6e1 100644
---
a/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/sink/committer/TestDorisCommitter.java
+++
b/flink-doris-connector/flink-doris-connector-base/src/test/java/org/apache/doris/flink/sink/committer/TestDorisCommitter.java
@@ -76,6 +76,9 @@ public class TestDorisCommitter {
httpResponse = mock(CloseableHttpResponse.class);
restServiceMockedStatic = mockStatic(RestService.class);
backendUtilMockedStatic = mockStatic(BackendUtil.class);
+ backendUtilMockedStatic
+ .when(() -> BackendUtil.getInstance(any(), any(), any()))
+ .thenCallRealMethod();
when(httpClient.execute(any())).thenReturn(httpResponse);
diff --git
a/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/sink/DorisSinkTest.java
b/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/sink/DorisSinkTest.java
index be28a9b2..2e52a409 100644
---
a/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/sink/DorisSinkTest.java
+++
b/flink-doris-connector/flink-doris-connector-flink1/src/test/java/org/apache/doris/flink/sink/DorisSinkTest.java
@@ -56,6 +56,9 @@ public class DorisSinkTest {
@Before
public void setUp() throws Exception {
backendUtilMockedStatic = mockStatic(BackendUtil.class);
+ backendUtilMockedStatic
+ .when(() -> BackendUtil.getInstance(any(), any(), any()))
+ .thenCallRealMethod();
backendUtilMockedStatic
.when(() -> BackendUtil.tryHttpConnection(any(), any()))
.thenReturn(true);
diff --git
a/flink-doris-connector/flink-doris-connector-flink2/src/test/java/org/apache/doris/flink/sink/DorisSinkTest.java
b/flink-doris-connector/flink-doris-connector-flink2/src/test/java/org/apache/doris/flink/sink/DorisSinkTest.java
index 8aaff8cc..0c08c6c9 100644
---
a/flink-doris-connector/flink-doris-connector-flink2/src/test/java/org/apache/doris/flink/sink/DorisSinkTest.java
+++
b/flink-doris-connector/flink-doris-connector-flink2/src/test/java/org/apache/doris/flink/sink/DorisSinkTest.java
@@ -58,6 +58,9 @@ public class DorisSinkTest {
@Before
public void setUp() throws Exception {
backendUtilMockedStatic = mockStatic(BackendUtil.class);
+ backendUtilMockedStatic
+ .when(() -> BackendUtil.getInstance(any(), any(), any()))
+ .thenCallRealMethod();
backendUtilMockedStatic
.when(() -> BackendUtil.tryHttpConnection(any(), any()))
.thenReturn(true);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
