What causes "No offset found for {tp} at timestamp {seek_to_timestamp_ms}" in xai-org/x-algorithm?

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

xai-org/x-algorithm · error · ValueError

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

Error message

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

What it means

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.

Source

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

        await seek_to_timestamp(consumer, assigned_partitions, seek_to_timestamp_ms)
    elif seek_to_offset is not None:
        await seek_consumer_to_offset(consumer, assigned_partitions, seek_to_offset)
    return assigned_partitions


async def seek_to_timestamp(
    consumer: AIOKafkaConsumer, assigned_partitions: list[TopicPartition], seek_to_timestamp_ms: int
):
    timestamps = {tp: seek_to_timestamp_ms for tp in assigned_partitions}
    offsets = await consumer.offsets_for_times(timestamps)

    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] = {}

View on GitHub (pinned to 24c60942c5)

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.

Example fix

# before
offset_and_ts = offsets.get(tp)
if offset_and_ts is not None and offset_and_ts.offset != -1:
    consumer.seek(tp, offset_and_ts.offset)
else:
    raise ValueError(f"No offset found for {tp} ...")

# after
offset_and_ts = offsets.get(tp)
if offset_and_ts is not None and offset_and_ts.offset != -1:
    consumer.seek(tp, offset_and_ts.offset)
else:
    rank_logger.warning(f"No offset at timestamp for {tp}; seeking to end")
    consumer.seek_to_end(tp)

Defensive patterns

Strategy: fallback

Validate before calling

offsets = await consumer.offsets_for_times(
    {tp: seek_to_timestamp_ms for tp in assigned_partitions}
)
missing = [tp for tp, o in offsets.items() if o is None or o.offset == -1]

Try / catch

try:
    await seek_to_timestamp(consumer, assigned_partitions, ts)
except ValueError:
    for tp in assigned_partitions:
        consumer.seek_to_end(tp)  # fallback position

Prevention

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.

When it happens

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

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

Related errors

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