apache/beam · critical · IOException

Exception while reading from Kafka

Error message

Exception while reading from Kafka

What it means

KafkaUnboundedReader wraps the exception thrown by its background Kafka consumer poll thread into an IOException with message 'Exception while reading from Kafka'. The reader polls Kafka on a dedicated thread; if that thread fails, the failure is stored in consumerPollException and rethrown on the pipeline thread when next() finds no records, so the beam runner fails the bundle.

Solutions

1. Inspect the cause attached to this IOException (consumerPollException) — the root broker/auth/deserialization error is there.
2. Verify bootstrap servers, security protocol, SASL/SSL config and ACLs from a worker node.
3. Check the topic/partitions still exist and the deserializer matches the topic's data format.
4. Add retry/failure handling at the pipeline level (runner retry policy) and Kafka client settings like session timeout to tolerate broker restarts.

try { reader.next(); } catch (IOException e) { Throwable root = e.getCause(); LOG.error("Kafka poll failed", root); /* retry or fail bundle */ }

When it happens

Trigger: The Kafka consumer poll thread fails — e.g. broker unreachable, authentication/authorization failure, invalid topic/partition, record deserialization error, timeout — and the reader subsequently calls next() before any records were dequeued.

Common situations: Kafka brokers down or DNS misconfigured in the cluster; SASL/SSL credentials expired or wrong; topic deleted or ACLs lacking DESCRIBE/READ; incompatible record deserializer throwing on a poisoned message.

Understand the failure class

Background: 'Something went wrong' / 'Request failed (500)' / 'HTTP error! status: 404' — what failed HTTP requests actually mean and how to find the real cause — this error's family across 28 libraries.

Related errors

• chunk send failed
• chunk send failed
• error fetching messages
• Failed to download file
• failed to receive header

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

Appendix: source

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

  private void nextBatch() throws IOException {
    curBatch = Collections.emptyIterator();

    ConsumerRecords<byte[], byte[]> records;
    try {
      // poll available records, wait (if necessary) up to the specified timeout.
      records =
          availableRecordsQueue.poll(recordsDequeuePollTimeout.getMillis(), TimeUnit.MILLISECONDS);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      LOG.warn("{}: Unexpected", this, e);
      return;
    }

    if (records == null) {
      // Check if the poll thread failed with an exception.
      if (consumerPollException.get() != null) {
        throw new IOException("Exception while reading from Kafka", consumerPollException.get());
      }
      if (recordsDequeuePollTimeout.isLongerThan(RECORDS_DEQUEUE_POLL_TIMEOUT_MIN)) {
        recordsDequeuePollTimeout = recordsDequeuePollTimeout.minus(Duration.millis(1));
        LOG.debug("Reducing poll timeout for reader to {}", recordsDequeuePollTimeout.getMillis());
      }
      return;
    }

    if (recordsDequeuePollTimeout.isShorterThan(RECORDS_DEQUEUE_POLL_TIMEOUT_MAX)) {
      recordsDequeuePollTimeout = recordsDequeuePollTimeout.plus(Duration.millis(1));
      LOG.debug("Increasing poll timeout for reader to {}", recordsDequeuePollTimeout.getMillis());
      LOG.debug("Record count: {}", records.count());
    }

    partitionStates.forEach(p -> p.recordIter = records.records(p.topicPartition).iterator());

    reportBacklog();
    reportBacklogMetrics();

View on GitHub (pinned to 12126d8942)