DataSQRL · velo · Sep 2, 2026
diff --git a/README.md b/README.md
@@ -236,6 +236,44 @@ Notes:
   container), so no extra Kubernetes capabilities are required. To use hardware `perf_events` instead, the pod
   needs `SYS_ADMIN` and a host `kernel.perf_event_paranoid` of `1` or lower.
 
+### Diagnosing Kafka connectivity
+
+The Flink Kafka enumerator collapses every pre-metadata failure into one opaque message:
+
+```
+org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: listNodes
+```
+
+DNS, a blocked port, a TLS mismatch and rejected credentials are indistinguishable in that line. The image ships
+`kafka-probe` to separate them, using the connector classes already on the image classpath, so it exercises the
+same client the job does:
+
+```bash
+kubectl exec -it <taskmanager-pod> -- kafka-probe
+kubectl exec -it <taskmanager-pod> -- kafka-probe broker-1:9096,broker-2:9096 my.topic
+```
+
+With no arguments it reads `KAFKA_BOOTSTRAP_SERVERS` and `KAFKA_PROBE_TOPIC`, and picks up
+`SQRL_KAFKA_SECURITY_PROTOCOL` (default `SASL_SSL`), `SQRL_KAFKA_SASL_MECHANISM` (default `SCRAM-SHA-512`),
+`SQRL_KAFKA_SASL_USERNAME` and `SQRL_KAFKA_SASL_PASSWORD` — so it tests the pod's real configuration rather than
+a hand-retyped copy of it. It reports DNS, TCP and TLS/SASL/metadata as separate stages and ends with a verdict:
+
+| Verdict | Meaning |
+|---------|---------|
+| `DNS` | Broker hostnames do not resolve |
+| `TCP` | Hostnames resolve, no port accepts a connection — security group, NACL, routing or peering |
+| `SSLException` | Wrong port for the listener, or an untrusted certificate chain |
+| `SaslAuthenticationException` | Credentials rejected |
+| `TopicAuthorizationException` | Authenticated, but the principal lacks ACLs on the topic |
+| `UnknownTopicOrPartitionException` | Authenticated and authorized, but the topic does not exist |
+| `REACHABLE` | The pod's configuration can reach Kafka |
+
+Set `KAFKA_PROBE_DEBUG=1` for the underlying `NetworkClient` and SASL handshake logs on stderr, without editing
+the cluster's log4j configuration. Note that a timeout during TCP means packets are being dropped rather than
+refused, which points at a firewall rather than an absent listener.
+
+`dig`, `nslookup`, `nc`, `ss`, `ip` and `unzip` are also installed for ad-hoc checks.
+
 ---
 
 ## Flink Extensions

diff --git a/flink-sql-runner/src/main/docker/Dockerfile b/flink-sql-runner/src/main/docker/Dockerfile
@@ -21,9 +21,16 @@ ARG TARGETARCH
 
 USER root
 
-# Install JDK 17 (replacing the default JRE)
+# Install JDK 17 (replacing the default JRE) alongside the network diagnostics needed to
+# triage connector connectivity from inside a running TaskManager: dnsutils (dig/nslookup),
+# netcat-openbsd (nc), iproute2 (ss/ip) and unzip.
 RUN apt-get update \
-    && apt-get install -y --no-install-recommends openjdk-17-jdk-headless \
+    && apt-get install -y --no-install-recommends \
+        openjdk-17-jdk-headless \
+        dnsutils \
+        netcat-openbsd \
+        iproute2 \
+        unzip \
     && ln -sf /usr/lib/jvm/java-17-openjdk-* /usr/lib/jvm/java-17-openjdk \
     && rm -rf /var/lib/apt/lists/*
 
@@ -55,6 +62,9 @@ COPY iceberg-aws-bundle-*.jar /opt/flink/lib
 COPY stdlib-utils-*.jar /opt/flink/lib
 COPY flink-sql-runner.uber.jar /opt/flink/lib/sql-runner.uber.jar
 COPY --chmod=755 sql-runner /opt/flink/bin/sql-runner
+COPY --chmod=755 kafka-probe /opt/flink/bin/kafka-probe
+COPY KafkaProbe.java /opt/flink/bin/KafkaProbe.java
+COPY kafka-probe-log4j2.properties /opt/flink/bin/kafka-probe-log4j2.properties
 COPY --chmod=755 entrypoint.sh /entrypoint.sh
 
 # noop.jar is an empty JAR. `flink run` requires a JAR positional argument, but the runner code

diff --git a/flink-sql-runner/src/main/docker/KafkaProbe.java b/flink-sql-runner/src/main/docker/KafkaProbe.java
@@ -0,0 +1,239 @@
+/*
+ * Copyright © 2026 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed 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.
+ */
+import java.io.IOException;
+import java.net.ConnectException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.net.UnknownHostException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.DescribeClusterOptions;
+import org.apache.kafka.clients.admin.DescribeTopicsOptions;
+
+/**
+ * Staged Kafka connectivity probe, run via the JDK single-file source launcher against the
+ * connector classes already on the image classpath.
+ *
+ * <p>The Flink enumerator reports every pre-metadata failure as the same opaque {@code Timed out
+ * waiting for a node assignment. Call: listNodes}. This separates that into DNS, TCP, and
+ * TLS/SASL/metadata so the failing layer is named rather than inferred.
+ */
+public final class KafkaProbe {
+
+  private static final int TCP_TIMEOUT_MS = 5_000;
+  private static final int ADMIN_TIMEOUT_MS = 15_000;
+
+  private KafkaProbe() {}
+
+  public static void main(String[] args) {
+    var bootstrap = arg(args, 0, env("KAFKA_BOOTSTRAP_SERVERS", ""));
+    var topic = arg(args, 1, env("KAFKA_PROBE_TOPIC", ""));
+
+    if (bootstrap.isBlank()) {
+      System.out.println(
+          "usage: kafka-probe [bootstrap.servers] [topic]   (defaults: $KAFKA_BOOTSTRAP_SERVERS, $KAFKA_PROBE_TOPIC)");
+      System.out.println("FAIL [config] bootstrap.servers is empty");
+      System.exit(2);
+    }
+
+    var protocol = env("SQRL_KAFKA_SECURITY_PROTOCOL", "SASL_SSL");
+    var mechanism = env("SQRL_KAFKA_SASL_MECHANISM", "SCRAM-SHA-512");
+    var user = env("SQRL_KAFKA_SASL_USERNAME", "");
+    var pass = env("SQRL_KAFKA_SASL_PASSWORD", "");
+
+    System.out.println("bootstrap.servers = " + bootstrap);
+    System.out.println("security.protocol = " + protocol);
+    System.out.println("sasl.mechanism    = " + mechanism);
+    System.out.println("sasl username     = " + (user.isBlank() ? "<EMPTY>" : user));
+    System.out.println(
+        "sasl password     = " + (pass.isBlank() ? "<EMPTY>" : "<set, " + pass.length() + " chars>"));
+    System.out.println();
+
+    var brokers = parse(bootstrap);
+    var dnsOk = stageDns(brokers);
+    var tcpOk = dnsOk && stageTcp(brokers);
+
+    if (!dnsOk) {
+      verdict("DNS", "broker hostnames do not resolve - check CoreDNS, the VPC resolver and any split-horizon zone");
+      System.exit(1);
+    }
+    if (!tcpOk) {
+      verdict("TCP", "hostnames resolve but no broker port accepts a connection - check security groups, NACLs, routing and VPC peering");
+      System.exit(1);
+    }
+    stageAdmin(bootstrap, protocol, mechanism, user, pass, topic);
+  }
+
+  private static boolean stageDns(List<Broker> brokers) {
+    System.out.println("== stage 1: DNS ==");
+    var ok = false;
+    for (var b : brokers) {
+      try {
+        var addrs = InetAddress.getAllByName(b.host());
+        var ips = new ArrayList<String>();
+        for (var a : addrs) {
+          ips.add(a.getHostAddress());
+        }
+        b.addresses().addAll(ips);
+        System.out.printf("  OK   %s -> %s%n", b.host(), String.join(", ", ips));
+        ok = true;
+      } catch (UnknownHostException e) {
+        System.out.printf("  FAIL %s -> unresolvable%n", b.host());
+      }
+    }
+    System.out.println();
+    return ok;
+  }
+
+  private static boolean stageTcp(List<Broker> brokers) {
+    System.out.println("== stage 2: TCP ==");
+    var ok = false;
+    for (var b : brokers) {
+      for (var ip : b.addresses()) {
+        var started = System.nanoTime();
+        try (var socket = new Socket()) {
+          socket.connect(new InetSocketAddress(ip, b.port()), TCP_TIMEOUT_MS);
+          System.out.printf("  OK   %s:%d open (%d ms)%n", ip, b.port(), elapsedMs(started));
+          ok = true;
+        } catch (SocketTimeoutException e) {
+          System.out.printf(
+              "  FAIL %s:%d timed out after %d ms - packets dropped, typically a security group or NACL%n",
+              ip, b.port(), elapsedMs(started));
+        } catch (ConnectException e) {
+          System.out.printf(
+              "  FAIL %s:%d refused - reachable but nothing is listening on that port%n", ip, b.port());
+        } catch (IOException e) {
+          System.out.printf("  FAIL %s:%d %s%n", ip, b.port(), e);
+        }
+      }
+    }
+    System.out.println();
+    return ok;
+  }
+
+  private static void stageAdmin(
+      String bootstrap, String protocol, String mechanism, String user, String pass, String topic) {
+    System.out.println("== stage 3: TLS + SASL + metadata ==");
+
+    var props = new Properties();
+    props.put("bootstrap.servers", bootstrap);
+    props.put("security.protocol", protocol);
+    props.put("request.timeout.ms", String.valueOf(ADMIN_TIMEOUT_MS));
+    props.put("default.api.timeout.ms", String.valueOf(ADMIN_TIMEOUT_MS));
+    if (protocol.startsWith("SASL")) {
+      props.put("sasl.mechanism", mechanism);
+      props.put(
+          "sasl.jaas.config",
+          "org.apache.kafka.common.security.scram.ScramLoginModule required username=\""
+              + user
+              + "\" password=\""
+              + pass
+              + "\";");
+    }
+
+    var timeout = new DescribeClusterOptions().timeoutMs(ADMIN_TIMEOUT_MS);
+    try (var admin = Admin.create(props)) {
+      var nodes = admin.describeCluster(timeout).nodes().get();
+      System.out.println("  OK   describeCluster -> " + nodes);
+
+      if (!topic.isBlank()) {
+        var described =
+            admin
+                .describeTopics(List.of(topic), new DescribeTopicsOptions().timeoutMs(ADMIN_TIMEOUT_MS))
+                .allTopicNames()
+                .get();
+        var partitions = described.get(topic).partitions().size();
+        System.out.printf("  OK   describeTopics -> %s (%d partitions)%n", topic, partitions);
+      }
+      System.out.println();
+      verdict("REACHABLE", "the connector configuration in this pod can reach Kafka");
+    } catch (Exception e) {
+      var root = rootCause(e);
+      var name = root.getClass().getSimpleName();
+      System.out.printf("  FAIL %s: %s%n%n", root.getClass().getName(), root.getMessage());
+      verdict(name, explain(name));
+      System.exit(1);
+    }
+  }
+
+  private static String explain(String exception) {
+    return switch (exception) {
+      case "TimeoutException" ->
+          "TCP opened but no broker completed a handshake - usually TLS interception or a broker refusing the listener";
+      case "SslAuthenticationException", "SSLHandshakeException", "SSLException" ->
+          "TLS failed - wrong port for the listener, or an untrusted certificate chain";
+      case "SaslAuthenticationException" -> "credentials rejected - check the SASL username and password";
+      case "TopicAuthorizationException" -> "authenticated, but the principal lacks ACLs on that topic";
+      case "GroupAuthorizationException" -> "authenticated, but the principal lacks ACLs on the consumer group";
+      case "UnknownTopicOrPartitionException" -> "authenticated and authorized, but the topic does not exist";
+      case "ConfigException" -> "client configuration rejected before any connection was attempted";
+      default -> "see the exception above";
+    };
+  }
+
+  private static void verdict(String label, String detail) {
+    System.out.println("VERDICT: " + label + " - " + detail);
+  }
+
+  private static List<Broker> parse(String bootstrap) {
+    var brokers = new ArrayList<Broker>();
+    for (var entry : bootstrap.split(",")) {
+      var trimmed = entry.trim();
+      if (trimmed.isEmpty()) {
+        continue;
+      }
+      var sep = trimmed.lastIndexOf(':');
+      if (sep < 0) {
+        brokers.add(new Broker(trimmed, 9092, new ArrayList<>()));
+      } else {
+        brokers.add(
+            new Broker(
+                trimmed.substring(0, sep),
+                Integer.parseInt(trimmed.substring(sep + 1)),
+                new ArrayList<>()));
+      }
+    }
+    return brokers;
+  }
+
+  private static Throwable rootCause(Throwable t) {
+    var root = t;
+    while (root.getCause() != null && root.getCause() != root) {
+      root = root.getCause();
+    }
+    return root;
+  }
+
+  private static long elapsedMs(long startedNanos) {
+    return Duration.ofNanos(System.nanoTime() - startedNanos).toMillis();
+  }
+
+  private static String arg(String[] args, int index, String fallback) {
+    return args.length > index ? args[index] : fallback;
+  }
+
+  private static String env(String name, String fallback) {
+    var value = System.getenv(name);
+    return value == null || value.isBlank() ? fallback : value;
+  }
+
+  private record Broker(String host, int port, List<String> addresses) {}
+}
diff --git a/flink-sql-runner/src/main/docker/kafka-probe b/flink-sql-runner/src/main/docker/kafka-probe
@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+#
+# Copyright © 2026 DataSQRL (contact@datasqrl.com)
+#
+# Licensed 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.
+#
+
+set -euo pipefail
+
+kafka_level=ERROR
+if [[ -n "${KAFKA_PROBE_DEBUG:-}" ]]; then
+  kafka_level=DEBUG
+fi
+
+exec "${JAVA_HOME:-/usr/lib/jvm/java-17-openjdk}/bin/java" \
+  -cp "/opt/flink/lib/*" \
+  -Dlog4j2.configurationFile=/opt/flink/bin/kafka-probe-log4j2.properties \
+  -DkafkaProbe.kafkaLevel="$kafka_level" \
+  /opt/flink/bin/KafkaProbe.java "$@"
diff --git a/flink-sql-runner/src/main/docker/kafka-probe-log4j2.properties b/flink-sql-runner/src/main/docker/kafka-probe-log4j2.properties
@@ -0,0 +1,31 @@
+#
+# Copyright © 2026 DataSQRL (contact@datasqrl.com)
+#
+# Licensed 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.
+#
+
+status = error
+name = kafka-probe
+
+# Client logs go to stderr so the staged probe output on stdout stays parseable.
+appender.console.type = Console
+appender.console.name = STDERR
+appender.console.target = SYSTEM_ERR
+appender.console.layout.type = PatternLayout
+appender.console.layout.pattern = %d{HH:mm:ss.SSS} %-5level %logger{1} - %msg%n
+
+rootLogger.level = ${sys:kafkaProbe.rootLevel:-ERROR}
+rootLogger.appenderRef.stderr.ref = STDERR
+
+logger.kafka.name = org.apache.kafka
+logger.kafka.level = ${sys:kafkaProbe.kafkaLevel:-ERROR}