Baike.dev
Log in
Back to tool/Back to issues
#28457·presto

presto-kafka connector bugs

Author: pratyakshsharmaCreated Sep 8, 2026Updated Sep 9, 2026
Labelsbug

A few issues have been reported with presto-kafka connector. Please find details below -

  1. KafkaSplitManager#getSplits() creates a new consumer instance by passing only one bootstrap server which is not recommended approach. Please see official kafka documentation - https://kafka.apache.org/43/configuration/consumer-configs/#consumerconfigs_bootstrap.servers. This essentially will revert the changes done as part of https://github.com/prestodb/presto/commit/79198697cc1847c2144aaee85f99ee8325069d6d.
  2. KafkaSplitManager#getSplits() checks if startTimestamp > endTimestamp. If true, IllegalArgumentException is thrown, but this if block never executes in reality since presto optimizer short circuits the query if there is no overlapping domain in the query. An example query is - select * from kafka.default.nation where _timestamp > 13 and _timestamp < 10;. So the check is dead and should be removed (though exception in point 4 below shows the same exception as the root cause, but after the fix was applied, it was not getting triggered)
  3. KafkaSplitManager#getSplits() creates kafka consumer instance for every call but never closes it.
  4. If a query uses where clause on _timestamp column such that only lower boundary (startTimestamp) is specified (where _timestamp > 1234), the query fails with below exception -
select * from kafka.tpch.nation where _timestamp > 6;

com.facebook.presto.spi.PrestoException: Cannot list splits for table 'nation' reading topic 'tpch.nation'
	at com.facebook.presto.kafka.KafkaSplitManager.getSplits(KafkaSplitManager.java:142)
	at com.facebook.presto.split.SplitManager.getSplits(SplitManager.java:89)
	at com.facebook.presto.split.CloseableSplitSourceProvider.getSplits(CloseableSplitSourceProvider.java:52)
	at com.facebook.presto.sql.planner.SplitSourceFactory$Visitor.lambda$visitTableScan$0(SplitSourceFactory.java:159)
	at com.facebook.presto.sql.planner.LazySplitSource.getDelegate(LazySplitSource.java:95)
	at com.facebook.presto.sql.planner.LazySplitSource.getConnectorId(LazySplitSource.java:47)
	at com.facebook.presto.execution.scheduler.SectionExecutionFactory.createStageScheduler(SectionExecutionFactory.java:299)
	at com.facebook.presto.execution.scheduler.SectionExecutionFactory.createStreamingLinkedStageExecutions(SectionExecutionFactory.java:258)
	at com.facebook.presto.execution.scheduler.SectionExecutionFactory.createStreamingLinkedStageExecutions(SectionExecutionFactory.java:235)
	at com.facebook.presto.execution.scheduler.SectionExecutionFactory.createSectionExecutions(SectionExecutionFactory.java:179)
	at com.facebook.presto.execution.scheduler.SqlQueryScheduler.createStageExecutions(SqlQueryScheduler.java:381)
	at com.facebook.presto.execution.scheduler.SqlQueryScheduler.<init>(SqlQueryScheduler.java:249)
	at com.facebook.presto.execution.scheduler.SqlQueryScheduler.createSqlQueryScheduler(SqlQueryScheduler.java:179)
	at com.facebook.presto.execution.SqlQueryExecution.createQueryScheduler(SqlQueryExecution.java:691)
	at com.facebook.presto.execution.SqlQueryExecution.lambda$start$4(SqlQueryExecution.java:507)
	at com.facebook.presto.common.RuntimeStats.lambda$recordWallAndCpuTime$9(RuntimeStats.java:171)
	at com.facebook.presto.common.RuntimeStats.recordWallAndCpuTime(RuntimeStats.java:158)
	at com.facebook.presto.common.RuntimeStats.recordWallAndCpuTime(RuntimeStats.java:170)
	at com.facebook.presto.execution.SqlQueryExecution.start(SqlQueryExecution.java:507)
	at com.facebook.presto.$gen.Presto_null__testversion____20260820_150357_1.run(Unknown Source)
	at com.facebook.presto.execution.SqlQueryManager.createQuery(SqlQueryManager.java:326)
	at com.facebook.presto.dispatcher.LocalDispatchQuery.lambda$startExecution$8(LocalDispatchQuery.java:217)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: java.lang.IllegalArgumentException: Invalid Kafka Offset start/end pair: 6 - 0
	at com.facebook.presto.kafka.KafkaSplitManager.getSplits(KafkaSplitManager.java:109)
	... 24 more
  1. NPE is thrown if _timestamp is specified with some value in where clause corresponding to which there is no offset present in kafka metadata. Example below -
select * from kafka.tpch.nation where _timestamp < 1787236457922;

com.facebook.presto.spi.PrestoException: Cannot list splits for table 'nation' reading topic 'tpch.nation'
	at com.facebook.presto.kafka.KafkaSplitManager.getSplits(KafkaSplitManager.java:142)
	at com.facebook.presto.split.SplitManager.getSplits(SplitManager.java:89)
	at com.facebook.presto.split.CloseableSplitSourceProvider.getSplits(CloseableSplitSourceProvider.java:52)
	at com.facebook.presto.sql.planner.SplitSourceFactory$Visitor.lambda$visitTableScan$0(SplitSourceFactory.java:159)
	at com.facebook.presto.sql.planner.LazySplitSource.getDelegate(LazySplitSource.java:95)
	at com.facebook.presto.sql.planner.LazySplitSource.getConnectorId(LazySplitSource.java:47)
	at com.facebook.presto.execution.scheduler.SectionExecutionFactory.createStageScheduler(SectionExecutionFactory.java:299)
	at com.facebook.presto.execution.scheduler.SectionExecutionFactory.createStreamingLinkedStageExecutions(SectionExecutionFactory.java:258)
	at com.facebook.presto.execution.scheduler.SectionExecutionFactory.createStreamingLinkedStageExecutions(SectionExecutionFactory.java:235)
	at com.facebook.presto.execution.scheduler.SectionExecutionFactory.createSectionExecutions(SectionExecutionFactory.java:179)
	at com.facebook.presto.execution.scheduler.SqlQueryScheduler.createStageExecutions(SqlQueryScheduler.java:381)
	at com.facebook.presto.execution.scheduler.SqlQueryScheduler.<init>(SqlQueryScheduler.java:249)
	at com.facebook.presto.execution.scheduler.SqlQueryScheduler.createSqlQueryScheduler(SqlQueryScheduler.java:179)
	at com.facebook.presto.execution.SqlQueryExecution.createQueryScheduler(SqlQueryExecution.java:691)
	at com.facebook.presto.execution.SqlQueryExecution.lambda$start$4(SqlQueryExecution.java:507)
	at com.facebook.presto.common.RuntimeStats.lambda$recordWallAndCpuTime$9(RuntimeStats.java:171)
	at com.facebook.presto.common.RuntimeStats.recordWallAndCpuTime(RuntimeStats.java:158)
	at com.facebook.presto.common.RuntimeStats.recordWallAndCpuTime(RuntimeStats.java:170)
	at com.facebook.presto.execution.SqlQueryExecution.start(SqlQueryExecution.java:507)
	at com.facebook.presto.$gen.Presto_null__testversion____20260820_150357_1.run(Unknown Source)
	at com.facebook.presto.execution.SqlQueryManager.createQuery(SqlQueryManager.java:326)
	at com.facebook.presto.dispatcher.LocalDispatchQuery.lambda$startExecution$8(LocalDispatchQuery.java:217)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: java.lang.NullPointerException: Cannot invoke "org.apache.kafka.clients.consumer.OffsetAndTimestamp.offset()" because "offsetAndTimestamp" is null
	at com.facebook.presto.kafka.KafkaSplitManager.findOffsetsByTimestamp(KafkaSplitManager.java:154)
	at com.facebook.presto.kafka.KafkaSplitManager.getSplits(KafkaSplitManager.java:120)
	... 24 more
  1. If kafka cluster is Sasl secured and the JAAS config password starts with a digit, the tokenizer used in kafka-clients library refuses to read the password properly resulting in below exception. This was reported here in OSS kafka - https://issues.apache.org/jira/browse/KAFKA-13352.
com.facebook.presto.spi.PrestoException: Cannot list splits for table 'quantum_production_job_events_eu_v1' reading topic 'quantum.production.job-events.v1'
	at com.facebook.presto.kafka.KafkaSplitManager.getSplits(KafkaSplitManager.java:142)
	at com.facebook.presto.split.SplitManager.getSplits(SplitManager.java:96)
	at com.facebook.presto.split.CloseableSplitSourceProvider.getSplits(CloseableSplitSourceProvider.java:62)
	at com.facebook.presto.sql.planner.SplitSourceFactory$Visitor.lambda$visitTableScan$0(SplitSourceFactory.java:166)
	at com.facebook.presto.sql.planner.LazySplitSource.getDelegate(LazySplitSource.java:95)
	at com.facebook.presto.sql.planner.LazySplitSource.getConnectorId(LazySplitSource.java:47)
	at com.facebook.presto.execution.scheduler.SectionExecutionFactory.createStageScheduler(SectionExecutionFactory.java:298)
	at com.facebook.presto.execution.scheduler.SectionExecutionFactory.createStreamingLinkedStageExecutions(SectionExecutionFactory.java:257)
	at com.facebook.presto.execution.scheduler.SectionExecutionFactory.createStreamingLinkedStageExecutions(SectionExecutionFactory.java:234)
	at com.facebook.presto.execution.scheduler.SectionExecutionFactory.createSectionExecutions(SectionExecutionFactory.java:178)
	at com.facebook.presto.execution.scheduler.SqlQueryScheduler.createStageExecutions(SqlQueryScheduler.java:380)
	at com.facebook.presto.execution.scheduler.SqlQueryScheduler.<init>(SqlQueryScheduler.java:248)
	at com.facebook.presto.execution.scheduler.SqlQueryScheduler.createSqlQueryScheduler(SqlQueryScheduler.java:178)
	at com.facebook.presto.execution.SqlQueryExecution.createQueryScheduler(SqlQueryExecution.java:789)
	at com.facebook.presto.execution.SqlQueryExecution.lambda$start$4(SqlQueryExecution.java:522)
	at com.facebook.presto.common.RuntimeStats.lambda$recordWallAndCpuTime$9(RuntimeStats.java:171)
	at com.facebook.presto.common.RuntimeStats.recordWallAndCpuTime(RuntimeStats.java:158)
	at com.facebook.presto.common.RuntimeStats.recordWallAndCpuTime(RuntimeStats.java:170)
	at com.facebook.presto.execution.SqlQueryExecution.start(SqlQueryExecution.java:522)
	at com.facebook.presto.$gen.Presto_0_299_SNAPSHOT_0b8f007__0_299_SNAPSHOT____20260820_184550_1.run(Unknown Source)
	at com.facebook.presto.execution.SqlQueryManager.createQuery(SqlQueryManager.java:331)
	at com.facebook.presto.dispatcher.LocalDispatchQuery.lambda$startExecution$8(LocalDispatchQuery.java:217)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:840)
Caused by: org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
	at org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer.<init>(ClassicKafkaConsumer.java:270)
	at org.apache.kafka.clients.consumer.internals.ConsumerDelegateCreator.create(ConsumerDelegateCreator.java:65)
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:600)
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:595)
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:576)
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:556)
	at com.facebook.presto.kafka.KafkaConsumerManager.createConsumer(KafkaConsumerManager.java:28)
	at com.facebook.presto.kafka.KafkaSplitManager.getSplits(KafkaSplitManager.java:94)
	... 24 more
Caused by: org.apache.kafka.common.KafkaException: Failed to create new NetworkClient
	at org.apache.kafka.clients.ClientUtils.createNetworkClient(ClientUtils.java:255)
	at org.apache.kafka.clients.ClientUtils.createNetworkClient(ClientUtils.java:163)
	at org.apache.kafka.clients.consumer.internals.ConsumerUtils.createConsumerNetworkClient(ConsumerUtils.java:90)
	at org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer.<init>(ClassicKafkaConsumer.java:197)
	... 31 more
Caused by: java.lang.IllegalArgumentException: Value not specified for key 'password' in JAAS config
	at org.apache.kafka.common.security.JaasConfig.parseAppConfigurationEntry(JaasConfig.java:116)
	at org.apache.kafka.common.security.JaasConfig.<init>(JaasConfig.java:63)
	at org.apache.kafka.common.security.JaasContext.load(JaasContext.java:95)
	at org.apache.kafka.common.security.JaasContext.loadClientContext(JaasContext.java:89)
	at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:168)
	at org.apache.kafka.common.network.ChannelBuilders.clientChannelBuilder(ChannelBuilders.java:82)
	at org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:120)
	at org.apache.kafka.clients.ClientUtils.createNetworkClient(ClientUtils.java:224)
	... 34 more

Your Environment

  • Presto version used: latest master, 0.299
  • Storage (HDFS/S3/GCS..):
  • Data source and connector used: kafka
  • Deployment (Cloud or On-prem):
  • Pastebin link to the complete debug logs:

Expected Behavior

Above issues are pretty basic ones and should not happen

Current Behavior

Above issues are happening.

Possible Solution

Steps to Reproduce

  1. setup kafka connector on local by following https://prestodb.io/docs/current/connector/kafka-tutorial.html.
  2. Modify server.properties file to include log.message.timestamp.type=LogAppendTime. This ensures timestamp is recorded with every message published in kafka. This is needed since tpch-kafka loader used in the tutorial uses older version of kafka which does not append timestamps with messages automatically. Without this property, _timestamp value is recorded as -1.
  3. Start kafka cluster, start presto server and test.

Screenshots (if appropriate)

Context

presto-kafka connector is not working properly.

Source: prestodb/presto

View original on GitHubView discussion on GitHub