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-kafka-connector.git
The following commit(s) were added to refs/heads/master by this push:
new 6e614d8 [improve] Cache available BE in BackendUtils to avoid
per-load HTTP probe (#99)
6e614d8 is described below
commit 6e614d8ea428f6cb0864f823f032fe01d59a4f99
Author: diggle <[email protected]>
AuthorDate: Wed Jul 29 10:04:21 2026 +0800
[improve] Cache available BE in BackendUtils to avoid per-load HTTP probe
(#99)
Cache available BE in BackendUtils to avoid per-load HTTP probe and shorten
probe timeout
---
.../doris/kafka/connector/utils/BackendUtils.java | 37 ++++++
.../connector/writer/commit/DorisCommitter.java | 2 +
.../writer/load/AsyncDorisStreamLoad.java | 1 +
.../connector/writer/load/DorisStreamLoad.java | 1 +
.../kafka/connector/utils/BackendUtilsTest.java | 134 +++++++++++++++++++++
5 files changed, 175 insertions(+)
diff --git
a/src/main/java/org/apache/doris/kafka/connector/utils/BackendUtils.java
b/src/main/java/org/apache/doris/kafka/connector/utils/BackendUtils.java
index c078b02..79ab85f 100644
--- a/src/main/java/org/apache/doris/kafka/connector/utils/BackendUtils.java
+++ b/src/main/java/org/apache/doris/kafka/connector/utils/BackendUtils.java
@@ -21,7 +21,9 @@ package org.apache.doris.kafka.connector.utils;
import java.net.HttpURLConnection;
import java.net.URL;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import org.apache.doris.kafka.connector.cfg.DorisOptions;
import org.apache.doris.kafka.connector.exception.DorisException;
import org.apache.doris.kafka.connector.model.BackendV2;
@@ -31,8 +33,14 @@ import org.slf4j.LoggerFactory;
public class BackendUtils {
private static final Logger LOG =
LoggerFactory.getLogger(BackendUtils.class);
+
+ /** TTL of a successful HTTP probe result for a BE (ms). */
+ private static final long PROBE_CACHE_TTL_MS = 5_000L;
+
private final List<BackendV2.BackendRowV2> backends;
private long pos;
+ /** backend -> last successful probe time (nanos). */
+ private final Map<String, Long> aliveProbeAtNanos = new HashMap<>();
public BackendUtils(List<BackendV2.BackendRowV2> backends) {
this.backends = backends;
@@ -43,18 +51,47 @@ public class BackendUtils {
return new BackendUtils(RestService.getBackendsV2(dorisOptions,
logger));
}
+ /**
+ * Pick a usable backend via round-robin so load is balanced across BEs. A
BE that was recently
+ * probed alive skips the HTTP probe within {@link #PROBE_CACHE_TTL_MS}.
+ */
public String getAvailableBackend() {
long tmp = pos + backends.size();
while (pos < tmp) {
BackendV2.BackendRowV2 backend = backends.get((int) (pos++ %
backends.size()));
String res = backend.toBackendString();
+ if (isRecentlyAlive(res)) {
+ return res;
+ }
if (tryHttpConnection(res)) {
+ aliveProbeAtNanos.put(res, System.nanoTime());
return res;
}
+ aliveProbeAtNanos.remove(res);
}
+ invalidateCache();
throw new DorisException("no available backend.");
}
+ /**
+ * Clear cached probe results. Callers should invoke this after a stream
load / commit failure
+ * so the next {@link #getAvailableBackend()} re-probes instead of
trusting a stale result.
+ */
+ public void invalidateCache() {
+ if (!aliveProbeAtNanos.isEmpty()) {
+ LOG.info("Invalidate doris backend probe cache, size={}",
aliveProbeAtNanos.size());
+ }
+ aliveProbeAtNanos.clear();
+ }
+
+ private boolean isRecentlyAlive(String backend) {
+ Long probedAt = aliveProbeAtNanos.get(backend);
+ if (probedAt == null) {
+ return false;
+ }
+ return (System.nanoTime() - probedAt) / 1_000_000L <
PROBE_CACHE_TTL_MS;
+ }
+
public static boolean tryHttpConnection(String backend) {
try {
backend = "http://"; + backend;
diff --git
a/src/main/java/org/apache/doris/kafka/connector/writer/commit/DorisCommitter.java
b/src/main/java/org/apache/doris/kafka/connector/writer/commit/DorisCommitter.java
index d9c0a7e..9968169 100644
---
a/src/main/java/org/apache/doris/kafka/connector/writer/commit/DorisCommitter.java
+++
b/src/main/java/org/apache/doris/kafka/connector/writer/commit/DorisCommitter.java
@@ -111,6 +111,7 @@ public class DorisCommitter {
"commit failed with {}, reason {}",
hostPort.get(),
reasonPhrase);
+ backendUtils.invalidateCache();
hostPort.set(backendUtils.getAvailableBackend());
throw new StreamLoadException(
"commit failed with {"
@@ -120,6 +121,7 @@ public class DorisCommitter {
+ "}");
} catch (Exception e) {
LOG.error("commit transaction failed, to retry,
{}", e.getMessage());
+ backendUtils.invalidateCache();
hostPort.set(backendUtils.getAvailableBackend());
throw new StreamLoadException("commit transaction
failed.", e);
}
diff --git
a/src/main/java/org/apache/doris/kafka/connector/writer/load/AsyncDorisStreamLoad.java
b/src/main/java/org/apache/doris/kafka/connector/writer/load/AsyncDorisStreamLoad.java
index a5987ad..21f6996 100644
---
a/src/main/java/org/apache/doris/kafka/connector/writer/load/AsyncDorisStreamLoad.java
+++
b/src/main/java/org/apache/doris/kafka/connector/writer/load/AsyncDorisStreamLoad.java
@@ -253,6 +253,7 @@ public class AsyncDorisStreamLoad extends DataLoad {
}
LOG.warn(err, ex);
+ backendUtils.invalidateCache();
throw new StreamLoadException(err, ex);
}
}
diff --git
a/src/main/java/org/apache/doris/kafka/connector/writer/load/DorisStreamLoad.java
b/src/main/java/org/apache/doris/kafka/connector/writer/load/DorisStreamLoad.java
index 6617c03..e4361be 100644
---
a/src/main/java/org/apache/doris/kafka/connector/writer/load/DorisStreamLoad.java
+++
b/src/main/java/org/apache/doris/kafka/connector/writer/load/DorisStreamLoad.java
@@ -135,6 +135,7 @@ public class DorisStreamLoad extends DataLoad {
}
LOG.warn(err, ex);
+ backendUtils.invalidateCache();
throw new StreamLoadException(err, ex);
}
}
diff --git
a/src/test/java/org/apache/doris/kafka/connector/utils/BackendUtilsTest.java
b/src/test/java/org/apache/doris/kafka/connector/utils/BackendUtilsTest.java
new file mode 100644
index 0000000..8639a18
--- /dev/null
+++ b/src/test/java/org/apache/doris/kafka/connector/utils/BackendUtilsTest.java
@@ -0,0 +1,134 @@
+/*
+ * 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.doris.kafka.connector.utils;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.util.Arrays;
+import java.util.Collections;
+import org.apache.doris.kafka.connector.exception.DorisException;
+import org.apache.doris.kafka.connector.model.BackendV2;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class BackendUtilsTest {
+
+ private ServerSocket alive;
+ private ServerSocket alive2;
+ private int alivePort;
+ private int alivePort2;
+
+ /**
+ * Port 1 on loopback is normally not bound and triggers an immediate
"connection refused", so
+ * the probe fails fast without waiting for the timeout.
+ */
+ private static final BackendV2.BackendRowV2 DEAD_BACKEND =
+ BackendV2.BackendRowV2.of("127.0.0.1", 1, true);
+
+ @Before
+ public void setUp() throws IOException {
+ alive = new ServerSocket();
+ alive.bind(new InetSocketAddress("127.0.0.1", 0));
+ alivePort = alive.getLocalPort();
+
+ alive2 = new ServerSocket();
+ alive2.bind(new InetSocketAddress("127.0.0.1", 0));
+ alivePort2 = alive2.getLocalPort();
+ }
+
+ @After
+ public void tearDown() throws IOException {
+ if (alive != null) {
+ alive.close();
+ }
+ if (alive2 != null) {
+ alive2.close();
+ }
+ }
+
+ private BackendV2.BackendRowV2 aliveBackend() {
+ return BackendV2.BackendRowV2.of("127.0.0.1", alivePort, true);
+ }
+
+ private BackendV2.BackendRowV2 aliveBackend2() {
+ return BackendV2.BackendRowV2.of("127.0.0.1", alivePort2, true);
+ }
+
+ @Test
+ public void testGetAvailableBackendReturnsAliveOne() {
+ BackendUtils utils = new BackendUtils(Arrays.asList(DEAD_BACKEND,
aliveBackend()));
+
+ String picked = utils.getAvailableBackend();
+ Assert.assertEquals("127.0.0.1:" + alivePort, picked);
+ }
+
+ @Test(expected = DorisException.class)
+ public void testGetAvailableBackendAllDead() {
+ BackendUtils utils = new
BackendUtils(Collections.singletonList(DEAD_BACKEND));
+ utils.getAvailableBackend();
+ }
+
+ @Test
+ public void testRoundRobinAcrossBackends() {
+ BackendUtils utils = new BackendUtils(Arrays.asList(aliveBackend(),
aliveBackend2()));
+
+ String first = utils.getAvailableBackend();
+ String second = utils.getAvailableBackend();
+
+ Assert.assertEquals("127.0.0.1:" + alivePort, first);
+ Assert.assertEquals("127.0.0.1:" + alivePort2, second);
+ Assert.assertNotEquals(first, second);
+ }
+
+ @Test
+ public void testProbeCacheHitSkipsProbe() throws IOException {
+ BackendUtils utils = new
BackendUtils(Collections.singletonList(aliveBackend()));
+
+ String first = utils.getAvailableBackend();
+
+ // Stop the only alive server. Within the probe-cache TTL the next
call must still return
+ // this backend without performing a fresh HTTP probe.
+ alive.close();
+
+ String second = utils.getAvailableBackend();
+ Assert.assertEquals(first, second);
+ }
+
+ @Test
+ public void testInvalidateCacheForcesReProbe() throws IOException {
+ BackendUtils utils = new
BackendUtils(Collections.singletonList(aliveBackend()));
+
+ String first = utils.getAvailableBackend();
+ Assert.assertNotNull(first);
+
+ alive.close();
+ utils.invalidateCache();
+
+ try {
+ utils.getAvailableBackend();
+ Assert.fail("expected DorisException after the only backend went
away");
+ } catch (DorisException expected) {
+ // ok
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
