xai-org/x-algorithm · error · ValueError

total_partitions must be >= 0, got

Error message

total_partitions must be >= 0, got {total_partitions}

What it means

Validation in _range_partitions: total_partitions (the partition count discovered for the topic) must be non-negative. A negative value indicates corrupted or nonsensical metadata being passed into the shard-range math, so the function refuses to compute a range.

Solutions

1. Check any manually configured partition count and set it to the real topic partition count or let discovery fill it in.
2. Log total_partitions right before the call to find where the negative value originates.
3. Fix the discovery/computation upstream so only discovered non-negative counts reach _range_partitions.

Example fix

# before
total_partitions = configured.get("partitions", -1)

# after
total_partitions = configured.get("partitions") or len(await discover_partition_count(consumer, topic))

assert isinstance(total_partitions, int) and total_partitions >= 0, total_partitions

When it happens

Trigger: handle_topic_offset or _consume_multi_consumer passing a negative partition count, typically from a bad len(partitions) path, an overridden total_partitions config set to -1, or arithmetic that underflowed.

Common situations: Manual override of partition count in config with -1 as 'auto/unset' sentinel; a code path computing total_partitions = something - something_else that went negative.

Related errors

• num_shards must be > 0, got
• No offset found for at timestamp
• Number of names must match number of dimensions
• Unable to create named shape with unnamed dimensions
• acks must be 0, 1, or 'all', got

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

Appendix: source

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

            available = await consumer.topics()
            raise ValueError(
                f"No partitions found for topic '{topic}'. Available topics: {available}"
            )
        count = len(partitions)
        rank_logger.info(f"Discovered {count} partitions for topic '{topic}' from Kafka metadata.")
        return count
    finally:
        try:
            await consumer.stop()
        except asyncio.CancelledError:
            pass


def _range_partitions(total_partitions: int, shard_index: int, num_shards: int) -> list[int]:
    if num_shards <= 0:
        raise ValueError(f"num_shards must be > 0, got {num_shards}")
    if total_partitions < 0:
        raise ValueError(f"total_partitions must be >= 0, got {total_partitions}")
    start = shard_index * total_partitions // num_shards
    end = (shard_index + 1) * total_partitions // num_shards
    return list(range(start, end))


async def handle_topic_offset(
    consumer: AIOKafkaConsumer,
    topic: str,
    shard_index: int,
    num_shards: int,
    reset_to_latest: bool,
    seek_to_timestamp_ms: int | None = None,
    seek_to_offset: dict[int, int] | None = None,
) -> list[TopicPartition]:
    assert seek_to_timestamp_ms is None or seek_to_offset is None, (
        "seek_to_timestamp_ms and seek_to_offset cannot both be set"
    )
    assert not (seek_to_timestamp_ms is not None and reset_to_latest), (

View on GitHub (pinned to 24c60942c5)