{"record":{"id":"8522d2f85dc37b06","repo":"prestodb/presto","slug":"generic-internal-error-8522d2","errorCode":"GENERIC_INTERNAL_ERROR","errorMessage":"Leader election in progress for Kafka topic '%s' partition %s","messagePattern":"Leader election in progress for Kafka topic '(.+?)' partition (.+?)","errorType":"error_code","errorClass":"PrestoException","httpStatus":null,"severity":"error","filePath":"presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaSplitManager.java","lineNumber":101,"sourceCode":"            ConnectorTransactionHandle transaction,\n            ConnectorSession session,\n            ConnectorTableLayoutHandle layout,\n            SplitSchedulingContext splitSchedulingContext)\n    {\n        KafkaTableHandle kafkaTableHandle = convertLayout(layout).getTable();\n        try {\n            String topic = kafkaTableHandle.getTopicName();\n            KafkaTableLayoutHandle layoutHandle = (KafkaTableLayoutHandle) layout;\n            HostAddress node = KafkaClusterMetadataHelper.selectRandom(clusterMetadataSupplier.getNodes(layoutHandle.getTable().getSchemaName()));\n\n            KafkaConsumer<ByteBuffer, ByteBuffer> consumer = consumerManager.createConsumer(Thread.currentThread().getName(), node);\n            List<PartitionInfo> partitions = consumer.partitionsFor(topic);\n            ImmutableList.Builder<ConnectorSplit> splits = ImmutableList.builder();\n\n            for (PartitionInfo partition : partitions) {\n                Node leader = partition.leader();\n                if (leader == null) {\n                    throw new PrestoException(GENERIC_INTERNAL_ERROR, format(\"Leader election in progress for Kafka topic '%s' partition %s\", topic, partition.partition()));\n                }\n\n                HostAddress partitionLeader = HostAddress.fromParts(leader.host(), leader.port());\n                long startTimestamp = layoutHandle.getStartOffsetTimestamp();\n                long endTimestamp = layoutHandle.getEndOffsetTimestamp();\n\n                if (startTimestamp > endTimestamp) {\n                    throw new IllegalArgumentException(String.format(\"Invalid Kafka Offset start/end pair: %s - %s\", startTimestamp, endTimestamp));\n                }\n\n                TopicPartition topicPartition = new TopicPartition(partition.topic(), partition.partition());\n                consumer.assign(ImmutableList.of(topicPartition));\n\n                long beginningOffset = (startTimestamp == 0) ?\n                        consumer.beginningOffsets(ImmutableList.of(topicPartition)).values().iterator().next() :\n                        findOffsetsByTimestamp(consumer, topicPartition, startTimestamp);\n                long endOffset = (endTimestamp == 0) ?\n                        consumer.endOffsets(ImmutableList.of(topicPartition)).values().iterator().next() :","sourceCodeStart":83,"sourceCodeEnd":119,"githubUrl":"https://github.com/prestodb/presto/blob/55bb57d202de3b926896fa966c2c4a44c779634e/presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaSplitManager.java#L83-L119","documentation":"While generating splits, KafkaSplitManager inspects each partition's leader via consumer.partitionsFor; if PartitionInfo.leader() is null, a leader election is in progress and the connector throws GENERIC_INTERNAL_ERROR. Presto needs the leader's host/port to build a split pinned to that broker, and cannot proceed until a leader exists.","triggerScenarios":"getSplits calls consumer.partitionsFor(topic) and iterates partitions; a partition whose replicas are all down or that is mid-leader-election returns leader()==null (e.g. right after broker restart or under-replicated topic).","commonSituations":"Querying a topic immediately after a broker crash/failover; topic created with replication factor 1 and that broker is down; topic auto-creation race where partitions have no leader yet.","solutions":["Wait for leader election to finish and re-run the query (usually seconds)","Check partition health: kafka-topics.sh --describe --topic <t> and confirm every partition has a Leader","Restore the down broker that hosts the leader replica","Increase replication factor so a single broker outage does not orphan partitions"],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// preflight: every partition has a leader\nkafka-topics.sh --bootstrap-server broker:9092 --describe --topic <topic> \\\n  | awk '$0 !~ /Leader: [^n]/ {print \"no leader:\", $0}'","typeGuard":"null","tryCatchPattern":"try { runQuery(); } catch (PrestoException e) {\n  if (e.getErrorCode() == GENERIC_INTERNAL_ERROR.toErrorCode() &&\n      e.getMessage().contains(\"Leader election in progress\")) {\n    Thread.sleep(5_000); retryQuery();\n  }\n}","preventionTips":["Use replication factor >= 3 for queried topics","Delay queries right after broker restarts","Alert on under-replicated / leaderless partitions"],"tags":["kafka","leader-election","broker","availability","presto-connector"],"backgroundTag":"kafka-no-leader-for-partition","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"}