apache/beam · error · IOException

%s: Timeout while initializing partition '%s'. Kafka client

Error message

%s: Timeout while initializing partition '%s'. Kafka client may not be able to connect to servers.

What it means

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.

Source

Thrown at sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java:131

    for (final PartitionState<K, V> pState : partitionStates) {
      Future<?> future = consumerPollThread.submit(() -> setupInitialOffset(pState));
      try {
        Duration timeout = resolveDefaultApiTimeout(spec);
        future.get(timeout.getMillis(), TimeUnit.MILLISECONDS);
      } catch (TimeoutException | ExecutionException e) {
        if (e instanceof TimeoutException
            || e.getCause() instanceof org.apache.kafka.common.errors.TimeoutException) {
          // TODO: Find out if manually waking up was only relevant for legacy Kafka clients.
          consumer.wakeup(); // This unblocks consumer stuck on network I/O.
          // Likely reason : Kafka servers are configured to advertise internal ips, but
          // those ips are not accessible from workers outside.
          String msg =
              String.format(
                  "%s: Timeout while initializing partition '%s'. "
                      + "Kafka client may not be able to connect to servers.",
                  this, pState.topicPartition);
          LOG.error("{}", msg);
          throw new IOException(msg);
        }
        throw new IOException(e);
      } catch (Exception e) {
        throw new IOException(e);
      }
      LOG.info(
          "{}: reading from {} starting at offset {}",
          name,
          pState.topicPartition,
          pState.nextOffset);
    }

    // Start consumer read loop.
    // Note that consumer is not thread safe, should not be accessed out side consumerPollLoop().
    consumerPollThread.submit(this::consumerPollLoop);

    // offsetConsumer setup :
    Map<String, Object> offsetConsumerConfig =

View on GitHub (pinned to 12126d8942)

Solutions

1. Verify bootstrap.servers and that it is reachable from the runner's workers (network/firewall/VPC routes)
2. Check Kafka broker health and that the topic/partition exists
3. Confirm SASL/SSL security settings match the broker configuration
4. 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

Example fix

// before
KafkaIO.<byte[], byte[]>read().withBootstrapServers("kafka-internal:9092")...
// after: use an address reachable from workers
KafkaIO.<byte[], byte[]>read().withBootstrapServers("kafka-public-broker:9092").withSecurityConfig(...)

// preflight connectivity from the worker environment
try (var socket = new java.net.Socket(host, port)) { /* reachable */ }

try {
  pipeline.run().waitUntilFinish();
} catch (IOException e) {
  if (e.getMessage().contains("Timeout while initializing partition")) {
    // retry with backoff after checking broker connectivity
  }
}

When it happens

Trigger: 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.

Common situations: 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.

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.

• Timeouts: ETIMEDOUT, deadlines, and hung requests — what actually expires when a request times out.

Related errors

• Timeout waiting for Python service startup after ${elapsed}
• Timeout waiting for the service {endpoint.getUrl()} to start
• Timed out waiting for service after ${timeoutMs}ms.
• Unable to verify project with ID ${projectId}
• Unable to get project number

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/a24d44198fce2274. Report an issue: GitHub.