{"record":{"id":"121af33411a2b121","repo":"prestodb/presto","slug":"kafka-consumer-error","errorCode":"KAFKA_CONSUMER_ERROR","errorMessage":"Failed to find offset by timestamp: %d for partition %d","messagePattern":"Failed to find offset by timestamp: (.+?) for partition (.+?)","errorType":"error_code","errorClass":"PrestoException","httpStatus":null,"severity":"error","filePath":"presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaSplitManager.java","lineNumber":157,"sourceCode":"            if (e instanceof PrestoException) {\n                throw e;\n            }\n            throw new PrestoException(KAFKA_SPLIT_ERROR, format(\"Cannot list splits for table '%s' reading topic '%s'\", kafkaTableHandle.getTableName(), kafkaTableHandle.getTopicName()), e);\n        }\n    }\n\n    private static long findOffsetsByTimestamp(KafkaConsumer<ByteBuffer, ByteBuffer> consumer, TopicPartition topicPartition, long timestamp)\n    {\n        try {\n            Map<TopicPartition, OffsetAndTimestamp> topicPartitionOffsets = consumer.offsetsForTimes(ImmutableMap.of(topicPartition, timestamp));\n            if (topicPartitionOffsets == null || topicPartitionOffsets.values().size() == 0) {\n                return 0;\n            }\n            OffsetAndTimestamp offsetAndTimestamp = topicPartitionOffsets.values().iterator().next();\n            return offsetAndTimestamp.offset();\n        }\n        catch (IllegalArgumentException e) {\n            throw new PrestoException(KAFKA_CONSUMER_ERROR, String.format(\"Failed to find offset by timestamp: %d for partition %d\", timestamp, topicPartition.partition()), e);\n        }\n    }\n\n    private static String readSchema(String dataSchemaLocation)\n    {\n        InputStream inputStream = null;\n        try {\n            if (isURI(dataSchemaLocation.trim().toLowerCase(ENGLISH))) {\n                try {\n                    inputStream = new URL(dataSchemaLocation).openStream();\n                }\n                catch (MalformedURLException e) {\n                    // try again before failing\n                    inputStream = new FileInputStream(dataSchemaLocation);\n                }\n            }\n            else {\n                inputStream = new FileInputStream(dataSchemaLocation);","sourceCodeStart":139,"sourceCodeEnd":175,"githubUrl":"https://github.com/prestodb/presto/blob/55bb57d202de3b926896fa966c2c4a44c779634e/presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaSplitManager.java#L139-L175","documentation":"findOffsetsByTimestamp calls KafkaConsumer.offsetsForTimes; when the Kafka client rejects the call with IllegalArgumentException the connector converts it into KAFKA_CONSUMER_ERROR with the timestamp and partition. Typically the timestamp is out of the accepted range or the partition/topic is invalid for this consumer.","triggerScenarios":"getSplits uses layout start/end timestamps -> findOffsetsByTimestamp -> offsetsForTimes; Kafka's offsetsForTimes throws IllegalArgumentException (e.g. negative timestamp, or timestamp handling unsupported given the broker's log.message.format/version).","commonSituations":"Negative or absurdly large epoch-millisecond timestamp in table properties; old Kafka brokers (<0.10.1) that don't support offsetsForTimes while the client does; mismatched kafka client/broker protocol versions.","solutions":["Set valid epoch-millisecond timestamps (0 < ts, within log retention) in the table's offset timestamp properties","Upgrade brokers or adjust the Kafka protocol version so offsetsForTimes is supported","Verify the TopicPartition passed actually belongs to the assigned topic"],"exampleFix":"// before\n\"startOffsetTimestamp\": -1\n// after\n\"startOffsetTimestamp\": 1699999200000","handlingStrategy":"validation","validationCode":"if (timestamp <= 0 || timestamp > System.currentTimeMillis())\n    throw new IllegalArgumentException(\"timestamp out of range: \" + timestamp);","typeGuard":"null","tryCatchPattern":"try { query(); } catch (PrestoException e) {\n  if (e.getErrorCode() == KAFKA_CONSUMER_ERROR.toErrorCode()) {\n    // correct timestamp values or broker version, retry\n  }\n}","preventionTips":["Use valid epoch-millisecond timestamps within retention","Ensure brokers >= 0.10.1 so offsetsForTimes is supported","Keep kafka client and broker protocol versions aligned"],"tags":["kafka","offsets","timestamps","presto-connector"],"backgroundTag":"kafka-offsets-for-times-failed","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"}