{"record":{"id":"46b26f837e75d89c","repo":"pathwaycom/pathway","slug":"topic-name-must-be-a-non-empty-string-got-an-em","errorCode":null,"errorMessage":"'topic_name' must be a non-empty string; got an empty string. Kafka does not allow empty topic names.","messagePattern":"'topic_name' must be a non-empty string; got an empty string\\. Kafka does not allow empty topic names\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/kafka/__init__.py","lineNumber":717,"sourceCode":"\n    >>> pw.io.kafka.write(\n    ...     t2,\n    ...     rdkafka_settings,\n    ...     \"test\",\n    ...     format=\"raw\",\n    ...     key=t2.bar,\n    ...     value=t2.foo,\n    ...     headers=[t2.baz],\n    ... )\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 producer can locate a broker; got \"\n            f\"{rdkafka_settings.get('bootstrap.servers')!r}.\"\n        )\n    if isinstance(topic_name, str) and not topic_name:\n        raise ValueError(\n            \"'topic_name' must be a non-empty string; got an empty string. \"\n            \"Kafka does not allow empty topic names.\"\n        )\n\n    output_format = MessageQueueOutputFormat.construct(\n        table,\n        format=format,\n        delimiter=delimiter,\n        key=key,\n        value=value,\n        headers=headers,\n        topic_name=topic_name if isinstance(topic_name, ColumnReference) else None,\n        schema_registry_settings=schema_registry_settings,\n        subject=subject,\n    )\n    output_table = output_format.table\n    remapped_sort_by = _remap_sort_by(sort_by, table, output_table)\n","sourceCodeStart":699,"sourceCodeEnd":735,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/kafka/__init__.py#L699-L735","documentation":"pw.io.kafka.write rejects an empty-string topic_name because Kafka does not allow zero-length topic names. Note this check only covers str inputs (isinstance(topic_name, str) and not topic_name) — a non-string topic_name is passed through and handled downstream, e.g. a ColumnReference is allowed for per-row topic routing.","triggerScenarios":"pw.io.kafka.write(table, rdkafka_settings, '', format='raw') — literal empty string; or topic_name computed from config that resolved to ''.","commonSituations":"Topic name from an unset env variable defaulting to ''; whitespace-only or accidentally cleared config value; templated deployment where the topic variable was not interpolated. Remember write uses 'topic_name' while read uses 'topic'.","solutions":["Pass a concrete topic name: pw.io.kafka.write(table, rdkafka_settings, 'test', format='raw').","Default config values at load time: topic_name = os.environ.get('KAFKA_TOPIC') or 'test'.","Kafka topic names must match [a-zA-Z0-9._-]{1,255} — validate user-supplied names against that pattern."],"exampleFix":"# before\ntopic_name = os.environ.get(\"KAFKA_OUTPUT_TOPIC\")  # unset -> None/''\npw.io.kafka.write(t, rdkafka_settings, topic_name or \"\", format=\"raw\")\n# after\ntopic_name = os.environ.get(\"KAFKA_OUTPUT_TOPIC\") or \"test\"\npw.io.kafka.write(t, rdkafka_settings, topic_name, format=\"raw\")","handlingStrategy":"validation","validationCode":"if isinstance(topic_name, str) and not topic_name:\n    raise SystemExit(\"topic_name must be a non-empty Kafka topic name\")\n\npw.io.kafka.write(table, rdkafka_settings, topic_name, format=\"raw\")","typeGuard":"def valid_write_topic(v) -> bool:\n    import re\n    if isinstance(v, str):\n        return bool(re.match(r\"^[a-zA-Z0-9._-]{1,255}$\", v))\n    return v is not None  # ColumnReference allowed for per-row routing","tryCatchPattern":null,"preventionTips":["Remember write uses 'topic_name' while read uses 'topic'.","Default output topics from env with a fail-fast check, never an empty string.","Validate against Kafka's topic charset before deployment."],"tags":["kafka","validation","configuration","pathway"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}