{"record":{"id":"a522d26d52e9f969","repo":"xai-org/x-algorithm","slug":"no-offset-found-for-tp-at-timestamp-seek-to-tim","errorCode":null,"errorMessage":"No offset found for {tp} at timestamp {seek_to_timestamp_ms}","messagePattern":"No offset found for (.+?) at timestamp (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"phoenix/xrex/data/streaming/kafkaconsumer.py","lineNumber":556,"sourceCode":"        await seek_to_timestamp(consumer, assigned_partitions, seek_to_timestamp_ms)\n    elif seek_to_offset is not None:\n        await seek_consumer_to_offset(consumer, assigned_partitions, seek_to_offset)\n    return assigned_partitions\n\n\nasync def seek_to_timestamp(\n    consumer: AIOKafkaConsumer, assigned_partitions: list[TopicPartition], seek_to_timestamp_ms: int\n):\n    timestamps = {tp: seek_to_timestamp_ms for tp in assigned_partitions}\n    offsets = await consumer.offsets_for_times(timestamps)\n\n    for tp in assigned_partitions:\n        offset_and_ts = offsets.get(tp)\n        if offset_and_ts is not None and offset_and_ts.offset != -1:\n            rank_logger.info(f\"Seeking to offset {offset_and_ts.offset} for {tp}\")\n            consumer.seek(tp, offset_and_ts.offset)\n        else:\n            raise ValueError(f\"No offset found for {tp} at timestamp {seek_to_timestamp_ms}\")\n\n\nasync def seek_consumer_to_offset(\n    consumer: AIOKafkaConsumer, assigned_partitions: list[TopicPartition], offsets: dict[int, int]\n):\n    for tp in assigned_partitions:\n        if tp.partition in offsets:\n            offset = offsets[tp.partition]\n            rank_logger.info(f\"Seeking to offset {offset} for {tp}\")\n            consumer.seek(tp, offset)\n        else:\n            raise ValueError(f\"No offset found for {tp} at timestamp {offsets}\")\n\n\nclass PartitionLagTracker:\n    def __init__(self) -> None:\n        self._partition_lag: dict[int, tuple[int, int]] = {}\n        self._latest_consumed_offsets: dict[int, int] = {}","sourceCodeStart":538,"sourceCodeEnd":574,"githubUrl":"https://github.com/xai-org/x-algorithm/blob/24c60942c5c5fdad3a6addffb4c6e6d2f228f04f/phoenix/xrex/data/streaming/kafkaconsumer.py#L538-L574","documentation":"seek_to_timestamp asks Kafka for offsets for a timestamp (offsets_for_times) and requires a valid offset for every assigned partition. If any TopicPartition returns None or offset == -1 (Kafka's sentinel for 'no offset at or after that timestamp'), a ValueError is raised because there is no meaningful position to seek to.","triggerScenarios":"seek_to_timestamp_ms is newer than the latest message in a partition (no data at/after that timestamp); timestamp older than log retention so all segments were deleted; partition empty; seeking to a future timestamp before new data is produced.","commonSituations":"Replay/start-from-timestamp logic using current wall-clock time on a low-traffic topic where the newest message is older than the requested timestamp; retention policy deleted old segments when seeking to an old timestamp; clock skew between producer and consumer hosts.","solutions":["Handle None/-1 per partition: for partitions with no offset, seek to end (or beginning) instead of failing, e.g. consumer.seek_to_end(tp) or seek(END).","Use a slightly older timestamp (e.g. now - 5 minutes) to ensure at least one record exists at/after it.","Verify the topic has data covering the requested window (kafka-run-class kafka.tools.GetOffsetShell --time <ts>).","Check retention settings if the timestamp is historical."],"exampleFix":"# before\noffset_and_ts = offsets.get(tp)\nif offset_and_ts is not None and offset_and_ts.offset != -1:\n    consumer.seek(tp, offset_and_ts.offset)\nelse:\n    raise ValueError(f\"No offset found for {tp} ...\")\n\n# after\noffset_and_ts = offsets.get(tp)\nif offset_and_ts is not None and offset_and_ts.offset != -1:\n    consumer.seek(tp, offset_and_ts.offset)\nelse:\n    rank_logger.warning(f\"No offset at timestamp for {tp}; seeking to end\")\n    consumer.seek_to_end(tp)","handlingStrategy":"fallback","validationCode":"offsets = await consumer.offsets_for_times(\n    {tp: seek_to_timestamp_ms for tp in assigned_partitions}\n)\nmissing = [tp for tp, o in offsets.items() if o is None or o.offset == -1]","typeGuard":null,"tryCatchPattern":"try:\n    await seek_to_timestamp(consumer, assigned_partitions, ts)\nexcept ValueError:\n    for tp in assigned_partitions:\n        consumer.seek_to_end(tp)  # fallback position","preventionTips":["Use a timestamp safely in the past relative to known data.","Handle None/-1 offsets per partition instead of failing the whole seek.","Check topic retention covers the seek window."],"tags":["kafka","offsets","timestamp","seek"],"backgroundTag":"kafka-offset-not-found","analyzedSha":"24c60942c5c5fdad3a6addffb4c6e6d2f228f04f","analyzedAt":"2026-08-28T11:40:14.686Z","schemaVersion":2},"datasetVersion":"2026-08-28T16:17:29.566Z"}