{"record":{"id":"10d3044b77c2386b","repo":"pathwaycom/pathway","slug":"the-mqtt-topic-to-read-from-must-not-be-empty","errorCode":null,"errorMessage":"The MQTT topic to read from must not be empty.","messagePattern":"The MQTT topic to read from must not be empty\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"python/pathway/io/mqtt/__init__.py","lineNumber":181,"sourceCode":"    >>> readings = pw.io.mqtt.read(\n    ...     \"mqtt://localhost:1883/?client_id=sensors-reader\",\n    ...     \"sensors/temperature\",\n    ...     format=\"json\",\n    ...     schema=SensorReading,\n    ... )\n    >>> deduplicated = readings.groupby(readings.event_id).reduce(\n    ...     readings.event_id,\n    ...     temperature=pw.reducers.any(readings.temperature),\n    ... )\n\n    The duplicates are exact copies of one message, so it does not matter which of\n    them ``pw.reducers.any`` picks. Which field to use as the identifier depends on\n    your data: a device-generated event id, a ``(device_id, timestamp)`` pair passed\n    to ``groupby`` as two columns, or any other combination that uniquely identifies\n    a message at the source.\n    \"\"\"\n    if topic == \"\":\n        raise ValueError(\"The MQTT topic to read from must not be empty.\")\n\n    data_storage = api.DataStorage(\n        storage_type=\"mqtt\",\n        path=uri,\n        topic=topic,\n        mode=api.ConnectorMode.STREAMING,\n        mqtt_settings=api.MqttSettings(\n            qos=qos,\n            retain=False,  # unused by reader\n        ),\n    )\n    schema, data_format = construct_schema_and_data_format(\n        \"binary\" if format == \"raw\" else format,\n        schema=schema,\n        csv_settings=None,\n        json_field_paths=json_field_paths,\n    )\n    data_source_options = datasource.DataSourceOptions(","sourceCodeStart":163,"sourceCodeEnd":199,"githubUrl":"https://github.com/pathwaycom/pathway/blob/fa2f74a4649b7c5908690cf60137263d8d80de5f/python/pathway/io/mqtt/__init__.py#L163-L199","documentation":"pw.io.mqtt.read subscribes to exactly one topic filter given by the topic argument; an empty string is not a valid MQTT topic filter, so the connector rejects it with this ValueError before building the storage. Note that wildcards (+ and #) ARE allowed here — only the empty string is refused.","triggerScenarios":"Calling pw.io.mqtt.read(uri, topic='') — usually because the topic comes from a config value, environment variable, or function argument that was never set.","commonSituations":"Reading topic from os.environ['MQTT_TOPIC'] when the var is unset and defaults to ''; a topic string built by concatenation where one part is missing; CLI argument forgotten.","solutions":["Provide a concrete topic or a valid wildcard filter, e.g. topic='sensors/+/temperature'","Validate configuration at startup: assert topic, 'MQTT topic must be configured'","Fail fast on missing env vars instead of defaulting to empty string"],"exampleFix":"# before\ntopic = os.getenv(\"MQTT_TOPIC\", \"\")\npw.io.mqtt.read(\"mqtt://localhost:1883\", topic=topic)\n\n# after\ntopic = os.environ[\"MQTT_TOPIC\"]  # KeyError points at the real misconfiguration\npw.io.mqtt.read(\"mqtt://localhost:1883\", topic=topic)","handlingStrategy":"validation","validationCode":"if not topic:\n    raise ValueError(\"MQTT read topic must be a non-empty topic filter\")\nassert \" \" not in topic  # MQTT forbids spaces in topics","typeGuard":"def is_valid_mqtt_filter(topic: str) -> bool:\n    return isinstance(topic, str) and len(topic) > 0 and \" \" not in topic and topic.startswith(\"$SYS\") is False or topic.startswith(\"$SYS/\")","tryCatchPattern":null,"preventionTips":["Make topic a required argument in wrappers instead of defaulting to ''","Use os.environ['MQTT_TOPIC'] (raises KeyError) rather than getenv with empty default","Validate topic strings in config loading before pipeline construction"],"tags":["mqtt","pathway","configuration","topic","validation"],"backgroundTag":null,"analyzedSha":"fa2f74a4649b7c5908690cf60137263d8d80de5f","analyzedAt":"2026-08-15T01:48:17.006Z","schemaVersion":2},"datasetVersion":"2026-08-15T17:31:12.345Z"}