diff --git a/README.md b/README.md index 965b1b2e..28a791f9 100644 --- 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 -- kafka-probe +kubectl exec -it -- 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 index 0d2815e0..26c1b97c 100644 --- 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 new file mode 100644 index 00000000..efd426f8 --- /dev/null +++ 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. + * + *

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() ? "" : user)); + System.out.println( + "sasl password = " + (pass.isBlank() ? "" : "")); + 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 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(); + 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 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 parse(String bootstrap) { + var brokers = new ArrayList(); + 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 addresses) {} +} diff --git a/flink-sql-runner/src/main/docker/kafka-probe b/flink-sql-runner/src/main/docker/kafka-probe new file mode 100644 index 00000000..c257d18b --- /dev/null +++ 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 new file mode 100644 index 00000000..3a3646f6 --- /dev/null +++ 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}