{"record":{"id":"da3cf485a08025a7","repo":"apache/beam","slug":"there-are-no-messages-has-a-timestamp-that-is-greater-than","errorCode":null,"errorMessage":"There are no messages has a timestamp that is greater than or equals to the target time or the message format version in this partition is before 0.10.0, topicPartition is: ","messagePattern":"There are no messages has a timestamp that is greater than or equals to the target time or the message format version in this partition is before 0\\.10\\.0, topicPartition is: ","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConsumerSpEL.java","lineNumber":160,"sourceCode":"  }\n\n  /**\n   * Look up the offset for the given partition by timestamp. Throws RuntimeException if there are\n   * no messages later than timestamp or if this partition does not support timestamp based offset.\n   */\n  @SuppressWarnings(\"unchecked\")\n  public static long offsetForTime(\n      Consumer<?, ?> consumer, TopicPartition topicPartition, Instant time) {\n\n    checkArgument(hasOffsetsForTimes, \"This Kafka Client must support Consumer.OffsetsForTimes().\");\n\n    // 'value' in the map returned by offsetFoTime() is null if there is no offset for the time.\n    OffsetAndTimestamp offsetAndTimestamp =\n        Iterables.getOnlyElement(\n            consumer.offsetsForTimes(ImmutableMap.of(topicPartition, time.getMillis())).values());\n\n    if (offsetAndTimestamp == null) {\n      throw new RuntimeException(\n          \"There are no messages has a timestamp that is greater than or \"\n              + \"equals to the target time or the message format version in this partition is \"\n              + \"before 0.10.0, topicPartition is: \"\n              + topicPartition);\n    } else {\n      return offsetAndTimestamp.offset();\n    }\n  }\n}\n","sourceCodeStart":142,"sourceCodeEnd":170,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConsumerSpEL.java#L142-L170","documentation":"Thrown by KafkaIO's ConsumerSpEL.offsetForTime() when consumer.offsetsForTimes() returns no OffsetAndTimestamp for the requested partition. This happens when the partition contains no message at or after the target time, or when Kafka cannot translate timestamps to offsets because the partition's message format version is older than 0.10.0 (timestamps were introduced in 0.10).","triggerScenarios":"Calling ConsumerSpEL.offsetForTime(consumer, topicPartition, time) when: (1) all messages in the partition are older than the requested Instant; (2) the partition's log.message.format.version is below 0.10 so no timestamps exist; (3) offsetsForTimes returns a null value for that partition.","commonSituations":"Using startReadTime / setStartReadTime with a time beyond the end of the topic; reading old topics written by pre-0.10 Kafka producers; broker-side message format downgrade; typo in the requested timestamp (e.g. future epoch millis).","solutions":["Choose a startReadTime within the topic's actual time range (e.g. earliest available timestamp).","Verify the broker/topic message.format.version is >= 0.10 so timestamps are recorded.","Check the topicPartition is correct and actually contains messages.","If nulls are acceptable for the use case, call consumer.offsetsForTimes() directly and handle null instead of using this helper.","Catch RuntimeException and fall back to earliest/latest offset."],"exampleFix":"// before\nbuilder.setStartReadTime(Instant.parse(\"2020-01-01T00:00:00Z\")); // before any data\n// after\nInstant start = Instant.now().minus(Duration.ofHours(1)); // within topic retention\nbuilder.setStartReadTime(start);","handlingStrategy":"validation","validationCode":"Map<TopicPartition, OffsetAndTimestamp> r = consumer.offsetsForTimes(Map.of(tp, millis));\nif (r.get(tp) == null) throw new IllegalStateException(\"No offsets for time \" + millis + \" on \" + tp);","typeGuard":null,"tryCatchPattern":"try { return ConsumerSpEL.offsetForTime(consumer, tp, ts); } catch (RuntimeException e) { return consumer beginningOffsets...; }","preventionTips":["Validate message format version >= 0.10","Choose timestamps inside retention"],"tags":["kafka","offset-lookup","timestamp","beam-io"],"backgroundTag":"record-not-found","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}