Copilot commented on code in PR #99:
URL:
https://github.com/apache/doris-kafka-connector/pull/99#discussion_r3663616953
##########
src/main/java/org/apache/doris/kafka/connector/utils/BackendUtils.java:
##########
@@ -31,8 +34,14 @@
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;
+
Review Comment:
PR description/title mention a ~30s backend/probe cache window, but the
implementation uses 5s. Please reconcile this (either adjust the TTL here or
update the PR description/expectations), otherwise behavior won’t match what
reviewers/users read.
##########
src/main/java/org/apache/doris/kafka/connector/utils/BackendUtils.java:
##########
@@ -43,18 +52,47 @@ public static BackendUtils getInstance(DorisOptions
dorisOptions, Logger logger)
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;
Review Comment:
The PR aims to “shorten probe timeout”, but the HTTP probe connect timeout
is still 60s. A long connect timeout can stall stream-load/commit failover
paths when a BE IP is unroutable (no immediate RST). Consider reducing this to
a few seconds (and/or making it configurable/constant).
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
