{"record":{"id":"d4022e172101279d","repo":"xai-org/x-algorithm","slug":"no-offset-found-for-tp-at-timestamp-offsets","errorCode":null,"errorMessage":"No offset found for {tp} at timestamp {offsets}","messagePattern":"No offset found for (.+?) at timestamp (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"phoenix/xrex/data/streaming/kafkaconsumer.py","lineNumber":568,"sourceCode":"    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] = {}\n\n    def record_consumed_offsets(self, messages: dict[TopicPartition, list[ConsumerRecord]]) -> None:\n        for tp, msg_list in messages.items():\n            if msg_list:\n                max_offset = max(msg.offset for msg in msg_list)\n                prev = self._latest_consumed_offsets.get(tp.partition)\n                if prev is None or max_offset > prev:\n                    self._latest_consumed_offsets[tp.partition] = max_offset\n\n    def update_partition_lag(self, partition: int, latest_offset: int, end_offset: int) -> None:\n        self._partition_lag[partition] = (latest_offset, end_offset)\n","sourceCodeStart":550,"sourceCodeEnd":586,"githubUrl":"https://github.com/xai-org/x-algorithm/blob/24c60942c5c5fdad3a6addffb4c6e6d2f228f04f/phoenix/xrex/data/streaming/kafkaconsumer.py#L550-L586","documentation":"seek_consumer_to_offset applies an explicit per-partition offset map and raises when an assigned partition has no entry in the offsets dict. Kafka's seek() requires a target offset for every partition being positioned, so a missing key is a caller error, not a broker condition.","triggerScenarios":"handle_topic_offset building the offsets dict from a subset of partitions (e.g. only partitions with committed offsets) while the consumer is assigned all partitions in the shard; partition count grew (topic expanded) and new partitions have no entry; shard range changed between discovery and assignment.","commonSituations":"See trigger scenarios.","solutions":["Before calling, intersect assigned_partitions with offsets.keys() or supply a default (e.g. OFFSET_END / OFFSET_BEGINNING) for partitions missing from the map.","If the topic was recently expanded, wait for rebalance/metadata to settle or rebuild the offsets map for the new partition count.","Log tp.partition and sorted(offsets) at call time to spot the mismatch quickly."],"exampleFix":"# before\nawait seek_consumer_to_offset(consumer, assigned_partitions, offsets)\n\n# after\nfrom kafka import OFFSET_END\nsafe_offsets = {p.partition: offsets.get(p.partition, OFFSET_END) for p in assigned_partitions}\nawait seek_consumer_to_offset(consumer, assigned_partitions, safe_offsets)","handlingStrategy":"type-guard","validationCode":"missing = [tp.partition for tp in assigned_partitions if tp.partition not in offsets]\nif missing:\n    raise ConfigError(f\"offsets map missing partitions: {missing}\")","typeGuard":"def offsets_cover_assignments(assigned: list[TopicPartition], offsets: dict[int, int]) -> bool:\n    return all(tp.partition in offsets for tp in assigned)","tryCatchPattern":null,"preventionTips":["Build the offsets map from the same partition set used for assignment.","Default missing partitions to OFFSET_END/OFFSET_BEGINNING explicitly."],"tags":["kafka","offsets","seek","validation"],"backgroundTag":"kafka-offset-not-found","analyzedSha":"24c60942c5c5fdad3a6addffb4c6e6d2f228f04f","analyzedAt":"2026-08-28T11:40:14.686Z","schemaVersion":2},"datasetVersion":"2026-08-28T16:17:29.566Z"}