{"record":{"id":"a445ffa29068efb7","repo":"apache/druid","slug":"failed-to-connect-to-kafka-in-sufficient-time","errorCode":null,"errorMessage":"Failed to connect to kafka in sufficient time","messagePattern":"Failed to connect to kafka in sufficient time","errorType":"exception","errorClass":"java.util.concurrent.TimeoutException","httpStatus":null,"severity":"error","filePath":"extensions-core/kafka-extraction-namespace/src/main/java/org/apache/druid/query/lookup/KafkaLookupExtractorFactory.java","lineNumber":248,"sourceCode":"              if (t instanceof CancellationException) {\n                LOG.debug(\"Topic [%s] cancelled\", topic);\n              } else {\n                LOG.error(t, \"Error in listening to [%s]\", topic);\n              }\n            }\n          },\n          Execs.directExecutor()\n      );\n      this.future = future;\n      final Stopwatch stopwatch = Stopwatch.createStarted();\n      try {\n        while (!startingReads.await(100, TimeUnit.MILLISECONDS) && connectTimeout > 0L) {\n          // Don't return until we have actually connected\n          if (future.isDone()) {\n            future.get();\n          } else {\n            if (stopwatch.elapsed(TimeUnit.MILLISECONDS) > connectTimeout) {\n              throw new TimeoutException(\"Failed to connect to kafka in sufficient time\");\n            }\n          }\n        }\n      }\n      catch (InterruptedException | ExecutionException | TimeoutException e) {\n        executorService.shutdown();\n        future.cancel(true);\n        LOG.error(e, \"Failed to start kafka extraction factory\");\n        cacheHandler.close();\n        return false;\n      }\n\n      started.set(true);\n      return true;\n    }\n  }\n\n  @Override","sourceCodeStart":230,"sourceCodeEnd":266,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-core/kafka-extraction-namespace/src/main/java/org/apache/druid/query/lookup/KafkaLookupExtractorFactory.java#L230-L266","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Verify bootstrap.servers in kafkaProperties and network reachability (telnet/nc to broker host:port)","Increase kafkaConnectTimeout in the lookup spec","Check broker-side issues (broker down, overload, authentication/ACL failures) in Kafka logs","Confirm any SASL/SSL properties are correct so the connection can complete"],"exampleFix":"// before\n\"kafkaProperties\": {\"bootstrap.servers\": \"broker1:9092\"}, \"kafkaConnectTimeout\": 1000\n// after\n\"kafkaProperties\": {\"bootstrap.servers\": \"broker1:9092\"}, \"kafkaConnectTimeout\": 30000","handlingStrategy":"try-catch","validationCode":"// verify brokers reachable before start()\nString servers = kafkaProperties.get(\"bootstrap.servers\");\nfor (String s : servers.split(\",\")) {\n  String[] hp = s.replaceFirst(\"^.*://\", \"\").split(\":\");\n  try (java.net.Socket sock = new java.net.Socket()) {\n    sock.connect(new java.net.InetSocketAddress(hp[0], Integer.parseInt(hp[1])), 3000);\n  } catch (IOException e) { throw new IllegalStateException(\"Unreachable broker: \" + s, e); }\n}","typeGuard":null,"tryCatchPattern":"try {\n  factory.start();\n} catch (RuntimeException e) {\n  if (e.getCause() instanceof TimeoutException) {\n    // back off and retry with larger connectTimeout / check brokers\n  } else { throw e; }\n}","preventionTips":["Pre-check broker connectivity from the Druid host","Set kafkaConnectTimeout generously (e.g. 30s+)","Validate SASL/SSL properties before deployment","Monitor Kafka broker health"],"tags":["kafka","timeout","connectivity"],"backgroundTag":"request-timeout","analyzedSha":"9b90983fd291f26935af934383ce360473179e4d","analyzedAt":"2026-09-07T13:32:30.957Z","contentChangedAt":"2026-09-07T13:32:30.957Z","schemaVersion":2},"datasetVersion":"2026-09-17T15:17:12.973Z"}