xai-org/x-algorithm · error · ValueError

No offset found for {tp} at timestamp {offsets}

Error message

No offset found for {tp} at timestamp {offsets}

What it means

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.

Source

Thrown at phoenix/xrex/data/streaming/kafkaconsumer.py:568

    for tp in assigned_partitions:
        offset_and_ts = offsets.get(tp)
        if offset_and_ts is not None and offset_and_ts.offset != -1:
            rank_logger.info(f"Seeking to offset {offset_and_ts.offset} for {tp}")
            consumer.seek(tp, offset_and_ts.offset)
        else:
            raise ValueError(f"No offset found for {tp} at timestamp {seek_to_timestamp_ms}")


async def seek_consumer_to_offset(
    consumer: AIOKafkaConsumer, assigned_partitions: list[TopicPartition], offsets: dict[int, int]
):
    for tp in assigned_partitions:
        if tp.partition in offsets:
            offset = offsets[tp.partition]
            rank_logger.info(f"Seeking to offset {offset} for {tp}")
            consumer.seek(tp, offset)
        else:
            raise ValueError(f"No offset found for {tp} at timestamp {offsets}")


class PartitionLagTracker:
    def __init__(self) -> None:
        self._partition_lag: dict[int, tuple[int, int]] = {}
        self._latest_consumed_offsets: dict[int, int] = {}

    def record_consumed_offsets(self, messages: dict[TopicPartition, list[ConsumerRecord]]) -> None:
        for tp, msg_list in messages.items():
            if msg_list:
                max_offset = max(msg.offset for msg in msg_list)
                prev = self._latest_consumed_offsets.get(tp.partition)
                if prev is None or max_offset > prev:
                    self._latest_consumed_offsets[tp.partition] = max_offset

    def update_partition_lag(self, partition: int, latest_offset: int, end_offset: int) -> None:
        self._partition_lag[partition] = (latest_offset, end_offset)

View on GitHub (pinned to 24c60942c5)

Solutions

1. Before calling, intersect assigned_partitions with offsets.keys() or supply a default (e.g. OFFSET_END / OFFSET_BEGINNING) for partitions missing from the map.
2. If the topic was recently expanded, wait for rebalance/metadata to settle or rebuild the offsets map for the new partition count.
3. Log tp.partition and sorted(offsets) at call time to spot the mismatch quickly.

Example fix

# before
await seek_consumer_to_offset(consumer, assigned_partitions, offsets)

# after
from kafka import OFFSET_END
safe_offsets = {p.partition: offsets.get(p.partition, OFFSET_END) for p in assigned_partitions}
await seek_consumer_to_offset(consumer, assigned_partitions, safe_offsets)

missing = [tp.partition for tp in assigned_partitions if tp.partition not in offsets]
if missing:
    raise ConfigError(f"offsets map missing partitions: {missing}")

def offsets_cover_assignments(assigned: list[TopicPartition], offsets: dict[int, int]) -> bool:
    return all(tp.partition in offsets for tp in assigned)

When it happens

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

Common situations: See trigger scenarios.

Related errors

• No offset found for {tp} at timestamp {seek_to_timestamp_ms}
• num_shards must be > 0, got {num_shards}
• total_partitions must be >= 0, got {total_partitions}
• type checking expression %s failed: invalid argument type: %
• Non-optional parameter %s must be declared before optional p

AI-assisted analysis of xai-org/x-algorithm@24c60942c5 (2026-08-28). Data as JSON: /api/errors/d4022e172101279d. Report an issue: GitHub.