{"record":{"id":"9ad17af914f0586f","repo":"pathwaycom/pathway","slug":"topic-must-be-a-non-empty-string-got-topic-r","errorCode":null,"errorMessage":"'topic' must be a non-empty string; got {topic!r}. Kafka does not allow empty topic names.","messagePattern":"'topic' must be a non-empty string; got (.+?)\\. Kafka does not allow empty topic names\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/kafka/__init__.py","lineNumber":361,"sourceCode":"            raise TypeError(\n                f\"'topic_names' must be a str or a list of str; got \"\n                f\"{type(topic_names).__name__}\"\n            )\n    if isinstance(topic, list):\n        if not topic:\n            raise ValueError(\n                \"'topic' must be a non-empty string; got an empty list. \"\n                \"Kafka does not allow empty topic names.\"\n            )\n        warnings.warn(\n            \"'topic' should be a str, not list. First element will be used.\",\n            DeprecationWarning,\n            stacklevel=_stacklevel + 4,\n        )\n        topic = topic[0]\n\n    if not isinstance(topic, str) or not topic:\n        raise ValueError(\n            f\"'topic' must be a non-empty string; got {topic!r}. \"\n            \"Kafka does not allow empty topic names.\"\n        )\n\n    check_deprecated_kwargs(kwargs, [\"topic_names\"], stacklevel=_stacklevel + 4)\n\n    data_storage = api.DataStorage(\n        storage_type=\"kafka\",\n        rdkafka_settings=rdkafka_settings,\n        topic=topic,\n        parallel_readers=parallel_readers,\n        start_from_timestamp_ms=start_from_timestamp_ms,\n        mode=internal_connector_mode(mode),\n    )\n\n    # TODO: support case when the key is scalar and the value is json\n    schema, data_format = construct_schema_and_data_format(\n        format,","sourceCodeStart":343,"sourceCodeEnd":379,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/kafka/__init__.py#L343-L379","documentation":"After all alias handling and list unwrapping, pw.io.kafka.read requires the final topic value to be a non-empty string; anything else (None, empty string, an int, etc.) raises ValueError echoing the offending value. Kafka itself rejects invalid topic names, but Pathway's message is more actionable.","triggerScenarios":"pw.io.kafka.read(rdkafka_settings, topic=''), topic=None, or topic=123. This is the last topic check — it fires when the value survived the alias/list branches but is still not a usable string.","commonSituations":"Topic read from an env variable that is unset (yielding '' or None); passing a topic name containing only whitespace; passing a non-string identifier (int id from a config system) by mistake.","solutions":["Pass a concrete non-empty string: topic='test-topic'.","If the topic comes from config, default it: topic = os.environ.get('KAFKA_TOPIC') or 'test-topic' — and strip whitespace.","Kafka topic names also must match [a-zA-Z0-9._-]{1,255}; validate that pattern for user-supplied names."],"exampleFix":"# before\ntopic = os.environ.get(\"KAFKA_TOPIC\")  # unset -> None\nt = pw.io.kafka.read(rdkafka_settings, topic=topic)\n# after\ntopic = os.environ.get(\"KAFKA_TOPIC\") or \"test-topic\"\nt = pw.io.kafka.read(rdkafka_settings, topic=topic)","handlingStrategy":"validation","validationCode":"import re\nKAFKA_TOPIC_RE = re.compile(r\"^[a-zA-Z0-9._-]{1,255}$\")\n\ndef valid_kafka_topic(name) -> bool:\n    return isinstance(name, str) and bool(KAFKA_TOPIC_RE.match(name))\n\nassert valid_kafka_topic(topic), f\"bad Kafka topic: {topic!r}\"\nt = pw.io.kafka.read(rdkafka_settings, topic=topic)","typeGuard":"def valid_kafka_topic(name) -> bool:\n    import re\n    return isinstance(name, str) and bool(re.match(r\"^[a-zA-Z0-9._-]{1,255}$\", name))","tryCatchPattern":null,"preventionTips":["Default env-derived topics with 'or' and strip whitespace.","Validate topic names against Kafka's charset rule at config load.","Never pass None through — convert unset config to a hard failure first."],"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"}