diggle1 commented on code in PR #99:
URL:
https://github.com/apache/doris-kafka-connector/pull/99#discussion_r3663956637
##########
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:
Shortening the probe timeout is out of scope for this PR —
tryHttpConnection() is intentionally unchanged from master (still 60s). Happy
to follow up in a separate PR if we want a shorter / configurable timeout.
--
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]
