{"record":{"id":"9ca5d2304f4cc999","repo":"apache/beam","slug":"exception-while-reading-from-kafka","errorCode":null,"errorMessage":"Exception while reading from Kafka","messagePattern":"Exception while reading from Kafka","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"critical","filePath":"sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java","lineNumber":746,"sourceCode":"\n  private void nextBatch() throws IOException {\n    curBatch = Collections.emptyIterator();\n\n    ConsumerRecords<byte[], byte[]> records;\n    try {\n      // poll available records, wait (if necessary) up to the specified timeout.\n      records =\n          availableRecordsQueue.poll(recordsDequeuePollTimeout.getMillis(), TimeUnit.MILLISECONDS);\n    } catch (InterruptedException e) {\n      Thread.currentThread().interrupt();\n      LOG.warn(\"{}: Unexpected\", this, e);\n      return;\n    }\n\n    if (records == null) {\n      // Check if the poll thread failed with an exception.\n      if (consumerPollException.get() != null) {\n        throw new IOException(\"Exception while reading from Kafka\", consumerPollException.get());\n      }\n      if (recordsDequeuePollTimeout.isLongerThan(RECORDS_DEQUEUE_POLL_TIMEOUT_MIN)) {\n        recordsDequeuePollTimeout = recordsDequeuePollTimeout.minus(Duration.millis(1));\n        LOG.debug(\"Reducing poll timeout for reader to {}\", recordsDequeuePollTimeout.getMillis());\n      }\n      return;\n    }\n\n    if (recordsDequeuePollTimeout.isShorterThan(RECORDS_DEQUEUE_POLL_TIMEOUT_MAX)) {\n      recordsDequeuePollTimeout = recordsDequeuePollTimeout.plus(Duration.millis(1));\n      LOG.debug(\"Increasing poll timeout for reader to {}\", recordsDequeuePollTimeout.getMillis());\n      LOG.debug(\"Record count: {}\", records.count());\n    }\n\n    partitionStates.forEach(p -> p.recordIter = records.records(p.topicPartition).iterator());\n\n    reportBacklog();\n    reportBacklogMetrics();","sourceCodeStart":728,"sourceCodeEnd":764,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java#L728-L764","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Inspect the cause attached to this IOException (consumerPollException) — the root broker/auth/deserialization error is there.","Verify bootstrap servers, security protocol, SASL/SSL config and ACLs from a worker node.","Check the topic/partitions still exist and the deserializer matches the topic's data format.","Add retry/failure handling at the pipeline level (runner retry policy) and Kafka client settings like session timeout to tolerate broker restarts."],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"try { reader.next(); } catch (IOException e) { Throwable root = e.getCause(); LOG.error(\"Kafka poll failed\", root); /* retry or fail bundle */ }","preventionTips":["Monitor broker connectivity and consumer lag","Validate SASL/SSL credentials before deploy","Keep deserializers matched to topic data format","Configure reasonable timeouts and retry policies on the runner"],"tags":["kafka","io-exception","consumer-poll","network","streaming"],"backgroundTag":"http-request-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}