{"record":{"id":"fa9a7c3bcb0bffc5","repo":"pathwaycom/pathway","slug":"parallel-readers-must-be-positive-got-parallel","errorCode":null,"errorMessage":"'parallel_readers' must be positive; got {parallel_readers}.","messagePattern":"'parallel_readers' must be positive; got (.+?)\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/kafka/__init__.py","lineNumber":271,"sourceCode":"            \"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(\n            f\"'start_from_timestamp_ms' must be non-negative; got \"\n            f\"{start_from_timestamp_ms}. The value is a Unix timestamp in \"\n            f\"milliseconds — negative values are pre-epoch and not \"\n            f\"meaningful for Kafka.\"\n        )\n    if autocommit_duration_ms is not None and autocommit_duration_ms <= 0:\n        raise ValueError(\n            f\"'autocommit_duration_ms' must be positive; got \"\n            f\"{autocommit_duration_ms}. It is the maximum time between \"\n            f\"two commits and zero/negative values would prevent commits \"\n            f\"from happening.\"\n        )\n\n    # When 'start_from_timestamp_ms' is set, the engine seeks lazily after","sourceCodeStart":253,"sourceCodeEnd":289,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/kafka/__init__.py#L253-L289","documentation":"parallel_readers in pw.io.kafka.read controls how many parallel consumer instances share the topic's partitions; each reader needs at least one instance to make progress. Pathway raises ValueError for any explicit value <= 0 because zero or negative readers would consume nothing.","triggerScenarios":"pw.io.kafka.read(..., parallel_readers=0) or parallel_readers=-2. None (the default, meaning a single reader) is fine — only explicit non-positive integers raise.","commonSituations":"Setting parallel_readers=0 expecting it to mean 'automatic/serial'; deriving the count from an env var or partition count calculation that yields 0 on small topics.","solutions":["Use parallel_readers=1 for serial reading, or match your partition count (e.g. parallel_readers=3) for parallelism.","Omit the parameter entirely if one reader is enough.","If the value is computed (e.g. min(partitions, workers)), guard it with max(1, value)."],"exampleFix":"# before\nt = pw.io.kafka.read(rdkafka_settings, topic=\"t\", parallel_readers=0)\n# after\nt = pw.io.kafka.read(rdkafka_settings, topic=\"t\", parallel_readers=1)","handlingStrategy":"validation","validationCode":"parallel_readers = max(1, int(parallel_readers)) if parallel_readers is not None else None\nt = pw.io.kafka.read(rdkafka_settings, topic=\"t\", parallel_readers=parallel_readers)","typeGuard":"def valid_parallel_readers(v) -> bool:\n    return v is None or (isinstance(v, int) and v > 0)","tryCatchPattern":null,"preventionTips":["Match parallel_readers to the topic's partition count, minimum 1.","Clamp computed values with max(1, n).","Omit the parameter when a single reader suffices."],"tags":["kafka","validation","parallelism","pathway"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T17:31:12.345Z"}