{"record":{"id":"1e0952326280a9b0","repo":"pathwaycom/pathway","slug":"rdkafka-settings-must-contain-a-non-empty-group-i","errorCode":null,"errorMessage":"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}.","messagePattern":"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 (.+?)\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/kafka/__init__.py","lineNumber":257,"sourceCode":"    ...         \"pet_id\": \"/pet/identification/id\",\n    ...         \"pet_name\": \"/pet/name\",\n    ...         \"pet_height\": \"/pet/measurements/1\"\n    ...     },\n    ... )\n\n    Note that you would not need to provide the JSONPath for ``pet_id`` if it is\n    at the top level of the key JSON.\n    \"\"\"\n    # The data_storage is common to all kafka connectors\n\n    if not rdkafka_settings.get(\"bootstrap.servers\"):\n        raise ValueError(\n            \"rdkafka_settings must contain a non-empty 'bootstrap.servers' \"\n            \"entry so the consumer can locate a broker; got \"\n            f\"{rdkafka_settings.get('bootstrap.servers')!r}.\"\n        )\n    if not rdkafka_settings.get(\"group.id\"):\n        raise ValueError(\n            \"rdkafka_settings must contain a non-empty 'group.id' entry: \"\n            \"Pathway's Kafka reader uses 'subscribe' (not 'assign'), which \"\n            \"librdkafka refuses to perform without a configured consumer \"\n            f\"group id; got {rdkafka_settings.get('group.id')!r}.\"\n        )\n\n    if max_backlog_size is not None and max_backlog_size <= 0:\n        raise ValueError(\n            f\"'max_backlog_size' must be positive; got {max_backlog_size}. \"\n            f\"A non-positive value would prevent any entry from being \"\n            f\"processed and the reader would never make progress.\"\n        )\n    if parallel_readers is not None and parallel_readers <= 0:\n        raise ValueError(\n            f\"'parallel_readers' must be positive; got {parallel_readers}.\"\n        )\n    if start_from_timestamp_ms is not None and start_from_timestamp_ms < 0:\n        raise ValueError(","sourceCodeStart":239,"sourceCodeEnd":275,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/kafka/__init__.py#L239-L275","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Set a group id: rdkafka_settings['group.id'] = 'my-consumer-group'.","If you don't care about group semantics, generate one: 'group.id': str(uuid.uuid4()) (this is what simple_read does).","Remember the group id controls offset committing — keep it stable across restarts if you want to resume where you left off."],"exampleFix":"# before\nrdkafka_settings = {\"bootstrap.servers\": \"localhost:9092\"}\n# after\nrdkafka_settings = {\n    \"bootstrap.servers\": \"localhost:9092\",\n    \"group.id\": \"my-consumer-group\",\n}","handlingStrategy":"validation","validationCode":"def with_consumer_defaults(settings: dict) -> dict:\n    settings = dict(settings)\n    if not settings.get(\"group.id\"):\n        import uuid\n        settings[\"group.id\"] = f\"pathway-{uuid.uuid4()}\"\n    return settings\n\nrdkafka_settings = with_consumer_defaults(rdkafka_settings)","typeGuard":"def has_group_id(settings: dict) -> bool:\n    return bool(settings.get(\"group.id\"))","tryCatchPattern":null,"preventionTips":["Always set 'group.id' in reader settings; keep it stable to preserve offsets across restarts.","Use pw.io.kafka.simple_read when you don't care about group management.","Include group.id in config-file validation for streaming jobs."],"tags":["kafka","configuration","consumer-group","pathway"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T17:31:12.345Z"}