apache/druid · error · java.util.concurrent.TimeoutException

Failed to connect to kafka in sufficient time

Error message

Failed to connect to kafka in sufficient time

What it means

KafkaLookupExtractorFactory.start() spawns a background consumer and waits up to kafkaConnectTimeout for it to become ready; if the Kafka consumer future is not done within that window, a TimeoutException('Failed to connect to kafka in sufficient time') is thrown. This guards against lookups silently starting without a live connection to the brokers.

Solutions

1. Verify bootstrap.servers in kafkaProperties and network reachability (telnet/nc to broker host:port)
2. Increase kafkaConnectTimeout in the lookup spec
3. Check broker-side issues (broker down, overload, authentication/ACL failures) in Kafka logs
4. Confirm any SASL/SSL properties are correct so the connection can complete

Example fix

// before
"kafkaProperties": {"bootstrap.servers": "broker1:9092"}, "kafkaConnectTimeout": 1000
// after
"kafkaProperties": {"bootstrap.servers": "broker1:9092"}, "kafkaConnectTimeout": 30000

// verify brokers reachable before start()
String servers = kafkaProperties.get("bootstrap.servers");
for (String s : servers.split(",")) {
  String[] hp = s.replaceFirst("^.*://", "").split(":");
  try (java.net.Socket sock = new java.net.Socket()) {
    sock.connect(new java.net.InetSocketAddress(hp[0], Integer.parseInt(hp[1])), 3000);
  } catch (IOException e) { throw new IllegalStateException("Unreachable broker: " + s, e); }
}

try {
  factory.start();
} catch (RuntimeException e) {
  if (e.getCause() instanceof TimeoutException) {
    // back off and retry with larger connectTimeout / check brokers
  } else { throw e; }
}

When it happens

Trigger: Calling start() when Kafka brokers are unreachable/slow (wrong bootstrap.servers, network/firewall issues, broker overload, DNS problems) such that the consumer does not connect within the configured connectTimeout.

Common situations: Lookup configs pointing at staging brokers from prod, Kafka cluster down or restarting, security (SASL/SSL) misconfiguration preventing the handshake, timeout set too low for a large cluster.

Understand the failure class

Background: Request timed out: what client-side request timeouts mean across libraries (Request timed out, TIMED_OUT, APITimeoutError) — this error's family across 39 libraries.

Related errors

• Resume command was not accepted within 5 seconds
• Already shut down, not starting again
• Already started, not starting again
• Cache for node role [ ] could not be initialized before…
• Cache initialization for node role

AI-assisted analysis of apache/druid@9b90983fd2 (2026-09-07). Data as JSON: /api/errors/a445ffa29068efb7. Report an issue: GitHub.

Appendix: source

Thrown at extensions-core/kafka-extraction-namespace/src/main/java/org/apache/druid/query/lookup/KafkaLookupExtractorFactory.java:248

              if (t instanceof CancellationException) {
                LOG.debug("Topic [%s] cancelled", topic);
              } else {
                LOG.error(t, "Error in listening to [%s]", topic);
              }
            }
          },
          Execs.directExecutor()
      );
      this.future = future;
      final Stopwatch stopwatch = Stopwatch.createStarted();
      try {
        while (!startingReads.await(100, TimeUnit.MILLISECONDS) && connectTimeout > 0L) {
          // Don't return until we have actually connected
          if (future.isDone()) {
            future.get();
          } else {
            if (stopwatch.elapsed(TimeUnit.MILLISECONDS) > connectTimeout) {
              throw new TimeoutException("Failed to connect to kafka in sufficient time");
            }
          }
        }
      }
      catch (InterruptedException | ExecutionException | TimeoutException e) {
        executorService.shutdown();
        future.cancel(true);
        LOG.error(e, "Failed to start kafka extraction factory");
        cacheHandler.close();
        return false;
      }

      started.set(true);
      return true;
    }
  }

  @Override

View on GitHub (pinned to 9b90983fd2)