{"record":{"id":"ac3be5c818f020d7","repo":"prestodb/presto","slug":"kafka-split-error","errorCode":"KAFKA_SPLIT_ERROR","errorMessage":"Cannot read data from topic '%s', partition '%s', startOffset %s, endOffset %s, leader %s ","messagePattern":"Cannot read data from topic '(.+?)', partition '(.+?)', startOffset (.+?), endOffset (.+?), leader (.+?) ","errorType":"error_code","errorClass":"PrestoException","httpStatus":null,"severity":"error","filePath":"presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaRecordSet.java","lineNumber":328,"sourceCode":"                if (messageAndOffsetIterator == null) {\n                    String threadName = Thread.currentThread().getName();\n\n                    if (consumer == null) {\n                        consumer = consumerManager.createConsumer(threadName, split.getLeader());\n                    }\n\n                    TopicPartition topicPartition = new TopicPartition(split.getTopicName(), split.getPartitionId());\n                    consumer.assign(ImmutableList.of(topicPartition));\n                    consumer.seek(topicPartition, cursorOffset);\n                    ConsumerRecords<ByteBuffer, ByteBuffer> records = consumer.poll(POLL_TIMEOUT);\n                    messageAndOffsetIterator = records.records(topicPartition).iterator();\n                }\n            }\n            catch (Exception e) { // Catch all exceptions because Kafka library is written in scala and checked exceptions are not declared in method signature.\n                if (e instanceof PrestoException) {\n                    throw e;\n                }\n                throw new PrestoException(\n                        KAFKA_SPLIT_ERROR,\n                        format(\n                                \"Cannot read data from topic '%s', partition '%s', startOffset %s, endOffset %s, leader %s \",\n                                split.getTopicName(),\n                                split.getPartitionId(),\n                                split.getStart(),\n                                split.getEnd(),\n                                split.getLeader()),\n                        e);\n            }\n        }\n    }\n}\n","sourceCodeStart":310,"sourceCodeEnd":342,"githubUrl":"https://github.com/prestodb/presto/blob/55bb57d202de3b926896fa966c2c4a44c779634e/presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaRecordSet.java#L310-L342","documentation":"openFetchRequest wraps any exception from the Kafka consumer's fetch/send call (other than PrestoException) into KAFKA_SPLIT_ERROR, reporting the topic, partition, offset range and leader. This means Presto could not read data from the Kafka broker for this split — connectivity, broker availability, offset-range, or serialization issues inside the raw consumer API.","triggerScenarios":"KafkaRecordSet.advanceNextPosition -> openFetchRequest calls consumer.send/poll on a KafkaConsumer; the Kafka scala client throws (broker unreachable, unknown topic/partition, offset out of range, leader changed mid-scan).","commonSituations":"Broker down or firewall blocking broker port from workers; topic deleted/renamed between split generation and read; advertised.listeners misconfiguration so workers resolve broker hostnames they cannot reach; endOffset beyond log retention (records deleted).","solutions":["Verify brokers are reachable from Presto worker nodes (telnet broker:9092) and kafka.nodes config is correct","Check the topic/partition still exists and offsets are within the retained log (kafka-run-class kafka.tools.GetOffsetShell)","Fix advertised.listeners on the brokers so returned host:port are resolvable by workers","Retry the query if it was a transient leader election"],"exampleFix":"// before\nkafka.nodes=internal-broker:9092\n// after - resolvable from workers, multiple nodes\nkafka.nodes=broker1.internal:9092,broker2.internal:9092","handlingStrategy":"retry","validationCode":"// preflight: check topic reachability from worker\nkafka-topics.sh --bootstrap-server broker:9092 --describe --topic <topic>","typeGuard":"null","tryCatchPattern":"try { readSplit(); } catch (PrestoException e) {\n  if (e.getErrorCode() == KAFKA_SPLIT_ERROR.toErrorCode()) {\n    // verify broker connectivity and offset range, then retry query\n  }\n}","preventionTips":["Ensure kafka.nodes are resolvable from ALL workers","Fix broker advertised.listeners to worker-reachable host:port","Keep query offsets within log retention window","Monitor broker availability before scheduling heavy scans"],"tags":["kafka","network","broker","presto-connector"],"backgroundTag":"kafka-broker-unreachable","analyzedSha":"55bb57d202de3b926896fa966c2c4a44c779634e","analyzedAt":"2026-09-04T12:50:26.162Z","contentChangedAt":"2026-09-04T12:50:26.162Z","schemaVersion":2},"datasetVersion":"2026-09-11T21:17:09.523Z"}