pathwaycom/pathway · error · ValueError

rdkafka_settings must contain a non-empty 'group.id' entry:

Error message

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 group id; got {rdkafka_settings.get('group.id')!r}.

What it means

pw.io.kafka.read uses librdkafka's consumer group 'subscribe' mechanism rather than manual 'assign', and librdkafka refuses to subscribe without a configured consumer group id. Pathway therefore validates up front that rdkafka_settings contains a non-empty 'group.id' and raises ValueError with the found value otherwise.

Source

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

    ...         "pet_id": "/pet/identification/id",
    ...         "pet_name": "/pet/name",
    ...         "pet_height": "/pet/measurements/1"
    ...     },
    ... )

    Note that you would not need to provide the JSONPath for ``pet_id`` if it is
    at the top level of the key JSON.
    """
    # 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(

View on GitHub (pinned to fa2f74a464)

Solutions

1. Set a group id: rdkafka_settings['group.id'] = 'my-consumer-group'.
2. If you don't care about group semantics, generate one: 'group.id': str(uuid.uuid4()) (this is what simple_read does).
3. Remember the group id controls offset committing — keep it stable across restarts if you want to resume where you left off.

Example fix

# before
rdkafka_settings = {"bootstrap.servers": "localhost:9092"}
# after
rdkafka_settings = {
    "bootstrap.servers": "localhost:9092",
    "group.id": "my-consumer-group",
}

def with_consumer_defaults(settings: dict) -> dict:
    settings = dict(settings)
    if not settings.get("group.id"):
        import uuid
        settings["group.id"] = f"pathway-{uuid.uuid4()}"
    return settings

rdkafka_settings = with_consumer_defaults(rdkafka_settings)

def has_group_id(settings: dict) -> bool:
    return bool(settings.get("group.id"))

When it happens

Trigger: pw.io.kafka.read(rdkafka_settings={'bootstrap.servers': 'localhost:9092'}, topic='t') — 'group.id' missing or empty. Note that pw.io.kafka.simple_read sets a random uuid group id automatically, so this only affects direct read() calls.

Common situations: Assuming Pathway assigns partitions and thus needs no group; copying producer-side settings (which need no group.id) into the reader; empty group id from an unset environment variable.

Related errors

• Failed to install dependencies
• Column {api.TIME_PSEUDOCOLUMN} cannot contain negative times
• 'subject' must be a non-empty string; got an empty string. S
• rdkafka_settings must contain a non-empty 'bootstrap.servers
• Missing topic name specification

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