pathwaycom/pathway · error · ValueError

'max_backlog_size' must be positive; got {max_backlog_size}.

Error message

'max_backlog_size' must be positive; got {max_backlog_size}. A non-positive value would prevent any entry from being processed and the reader would never make progress.

What it means

In pw.io.kafka.read, max_backlog_size bounds how many unread Kafka entries may accumulate before the connector slows down reading. A value of zero or below would mean the reader can never buffer anything, so no entry could ever be processed and the pipeline would stall forever — Pathway rejects it up front with ValueError.

Source

Thrown at python/pathway/io/kafka/__init__.py:265

    """
    # The data_storage is common to all kafka connectors

    if not rdkafka_settings.get("bootstrap.servers"):
        raise ValueError(
            "rdkafka_settings must contain a non-empty 'bootstrap.servers' "
            "entry so the consumer can locate a broker; got "
            f"{rdkafka_settings.get('bootstrap.servers')!r}."
        )
    if not rdkafka_settings.get("group.id"):
        raise ValueError(
            "rdkafka_settings must contain a non-empty 'group.id' entry: "
            "Pathway's Kafka reader uses 'subscribe' (not 'assign'), which "
            "librdkafka refuses to perform without a configured consumer "
            f"group id; got {rdkafka_settings.get('group.id')!r}."
        )

    if max_backlog_size is not None and max_backlog_size <= 0:
        raise ValueError(
            f"'max_backlog_size' must be positive; got {max_backlog_size}. "
            f"A non-positive value would prevent any entry from being "
            f"processed and the reader would never make progress."
        )
    if parallel_readers is not None and parallel_readers <= 0:
        raise ValueError(
            f"'parallel_readers' must be positive; got {parallel_readers}."
        )
    if start_from_timestamp_ms is not None and start_from_timestamp_ms < 0:
        raise ValueError(
            f"'start_from_timestamp_ms' must be non-negative; got "
            f"{start_from_timestamp_ms}. The value is a Unix timestamp in "
            f"milliseconds — negative values are pre-epoch and not "
            f"meaningful for Kafka."
        )
    if autocommit_duration_ms is not None and autocommit_duration_ms <= 0:
        raise ValueError(
            f"'autocommit_duration_ms' must be positive; got "

View on GitHub (pinned to fa2f74a464)

Solutions

1. Use a positive value sized to your throughput, e.g. max_backlog_size=1000.
2. If you meant 'no limit', pass max_backlog_size=None (or omit the argument).
3. Validate the value at config-load time if it is computed dynamically.

Example fix

# before
t = pw.io.kafka.read(rdkafka_settings, topic="t", max_backlog_size=0)
# after
t = pw.io.kafka.read(rdkafka_settings, topic="t", max_backlog_size=1000)

if max_backlog_size is not None and max_backlog_size <= 0:
    raise SystemExit("max_backlog_size must be a positive integer or None")

t = pw.io.kafka.read(rdkafka_settings, topic="t", max_backlog_size=max_backlog_size)

def valid_backlog(v) -> bool:
    return v is None or (isinstance(v, int) and v > 0)

When it happens

Trigger: pw.io.kafka.read(..., max_backlog_size=0) or max_backlog_size=-5. Passing None is allowed (no limit) — only explicit non-positive integers raise.

Common situations: Setting max_backlog_size=0 intending 'no backlog' when the developer actually wants None (unbounded) or a small positive bound; computing the value from another config variable that can degenerate to 0.

Related errors

• Failed to install dependencies
• Column {pseudocolumn} has to contain integers only.
• Column {api.TIME_PSEUDOCOLUMN} cannot contain negative times
• Unsupported argument for {data_format!r} format: 'value'
• 'delimiter' is only meaningful for the 'dsv' format, but {fo

AI-assisted analysis of pathwaycom/pathway@fa2f74a464 (2026-08-15). Data as JSON: /api/errors/65e1297a245bf7ae. Report an issue: GitHub.