{"record":{"id":"a24d44198fce2274","repo":"apache/beam","slug":"s-timeout-while-initializing-partition-s-kafka-client-may","errorCode":null,"errorMessage":"%s: Timeout while initializing partition '%s'. Kafka client may not be able to connect to servers.","messagePattern":"(.+?): Timeout while initializing partition '(.+?)'\\. Kafka client may not be able to connect to servers\\.","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java","lineNumber":131,"sourceCode":"    for (final PartitionState<K, V> pState : partitionStates) {\n      Future<?> future = consumerPollThread.submit(() -> setupInitialOffset(pState));\n      try {\n        Duration timeout = resolveDefaultApiTimeout(spec);\n        future.get(timeout.getMillis(), TimeUnit.MILLISECONDS);\n      } catch (TimeoutException | ExecutionException e) {\n        if (e instanceof TimeoutException\n            || e.getCause() instanceof org.apache.kafka.common.errors.TimeoutException) {\n          // TODO: Find out if manually waking up was only relevant for legacy Kafka clients.\n          consumer.wakeup(); // This unblocks consumer stuck on network I/O.\n          // Likely reason : Kafka servers are configured to advertise internal ips, but\n          // those ips are not accessible from workers outside.\n          String msg =\n              String.format(\n                  \"%s: Timeout while initializing partition '%s'. \"\n                      + \"Kafka client may not be able to connect to servers.\",\n                  this, pState.topicPartition);\n          LOG.error(\"{}\", msg);\n          throw new IOException(msg);\n        }\n        throw new IOException(e);\n      } catch (Exception e) {\n        throw new IOException(e);\n      }\n      LOG.info(\n          \"{}: reading from {} starting at offset {}\",\n          name,\n          pState.topicPartition,\n          pState.nextOffset);\n    }\n\n    // Start consumer read loop.\n    // Note that consumer is not thread safe, should not be accessed out side consumerPollLoop().\n    consumerPollThread.submit(this::consumerPollLoop);\n\n    // offsetConsumer setup :\n    Map<String, Object> offsetConsumerConfig =","sourceCodeStart":113,"sourceCodeEnd":149,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java#L113-L149","documentation":"KafkaUnboundedReader.start() waits for the consumer to be assigned the partition and become ready; if initialization times out, it logs and throws an IOException stating the Kafka client may not be able to connect to servers. This is a startup connectivity/liveness guard.","triggerScenarios":"start() is invoked and the consumer fails to be assigned pState.topicPartition within the bootstrap timeout — brokers unreachable, wrong bootstrap servers, security/auth handshake hanging, DNS failure.","commonSituations":"Wrong bootstrap.servers in KafkaIO.read(), brokers behind a firewall/VPC boundary (Dataflow workers can't reach Kafka), SASL/SSL misconfiguration causing stalled handshakes, or brokers down.","solutions":["Verify bootstrap.servers and that it is reachable from the runner's workers (network/firewall/VPC routes)","Check Kafka broker health and that the topic/partition exists","Confirm SASL/SSL security settings match the broker configuration","Increase the consumer start timeout if the cluster is slow to respond, and inspect the logged 'Kafka client may not be able to connect to servers' context"],"exampleFix":"// before\nKafkaIO.<byte[], byte[]>read().withBootstrapServers(\"kafka-internal:9092\")...\n// after: use an address reachable from workers\nKafkaIO.<byte[], byte[]>read().withBootstrapServers(\"kafka-public-broker:9092\").withSecurityConfig(...)","handlingStrategy":"retry","validationCode":"// preflight connectivity from the worker environment\ntry (var socket = new java.net.Socket(host, port)) { /* reachable */ }","typeGuard":null,"tryCatchPattern":"try {\n  pipeline.run().waitUntilFinish();\n} catch (IOException e) {\n  if (e.getMessage().contains(\"Timeout while initializing partition\")) {\n    // retry with backoff after checking broker connectivity\n  }\n}","preventionTips":["Confirm bootstrap.servers resolve and are routable from runner workers","Open firewall/VPC routes to Kafka brokers (9092/SSL port)","Validate SASL/SSL configuration before launch","Monitor broker availability and DNS from the worker network"],"tags":["java","kafka","network","timeout","connectivity"],"backgroundTag":"request-timeout","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}