prestodb/presto · error · PrestoException

KAFKA_CONSUMER_ERROR

KAFKA_CONSUMER_ERROR

Error message

Failed to find offset by timestamp: %d for partition %d

What it means

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.

Source

Thrown at presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaSplitManager.java:157

            if (e instanceof PrestoException) {
                throw e;
            }
            throw new PrestoException(KAFKA_SPLIT_ERROR, format("Cannot list splits for table '%s' reading topic '%s'", kafkaTableHandle.getTableName(), kafkaTableHandle.getTopicName()), e);
        }
    }

    private static long findOffsetsByTimestamp(KafkaConsumer<ByteBuffer, ByteBuffer> consumer, TopicPartition topicPartition, long timestamp)
    {
        try {
            Map<TopicPartition, OffsetAndTimestamp> topicPartitionOffsets = consumer.offsetsForTimes(ImmutableMap.of(topicPartition, timestamp));
            if (topicPartitionOffsets == null || topicPartitionOffsets.values().size() == 0) {
                return 0;
            }
            OffsetAndTimestamp offsetAndTimestamp = topicPartitionOffsets.values().iterator().next();
            return offsetAndTimestamp.offset();
        }
        catch (IllegalArgumentException e) {
            throw new PrestoException(KAFKA_CONSUMER_ERROR, String.format("Failed to find offset by timestamp: %d for partition %d", timestamp, topicPartition.partition()), e);
        }
    }

    private static String readSchema(String dataSchemaLocation)
    {
        InputStream inputStream = null;
        try {
            if (isURI(dataSchemaLocation.trim().toLowerCase(ENGLISH))) {
                try {
                    inputStream = new URL(dataSchemaLocation).openStream();
                }
                catch (MalformedURLException e) {
                    // try again before failing
                    inputStream = new FileInputStream(dataSchemaLocation);
                }
            }
            else {
                inputStream = new FileInputStream(dataSchemaLocation);

View on GitHub (pinned to 55bb57d202)

Solutions

1. Set valid epoch-millisecond timestamps (0 < ts, within log retention) in the table's offset timestamp properties
2. Upgrade brokers or adjust the Kafka protocol version so offsetsForTimes is supported
3. Verify the TopicPartition passed actually belongs to the assigned topic

Example fix

// before
"startOffsetTimestamp": -1
// after
"startOffsetTimestamp": 1699999200000

if (timestamp <= 0 || timestamp > System.currentTimeMillis())
    throw new IllegalArgumentException("timestamp out of range: " + timestamp);

null

try { query(); } catch (PrestoException e) {
  if (e.getErrorCode() == KAFKA_CONSUMER_ERROR.toErrorCode()) {
    // correct timestamp values or broker version, retry
  }
}

When it happens

Trigger: 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).

Common situations: 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.

Related errors

• Invalid Kafka Offset start/end pair: %s - %s
• unexpected internal column '%s'
• KAFKA_SCHEMA_ERROR
• KAFKA_SPLIT_ERROR
• GENERIC_INTERNAL_ERROR

AI-assisted analysis of prestodb/presto@55bb57d202 (2026-09-04). Data as JSON: /api/errors/121af33411a2b121. Report an issue: GitHub.