pathwaycom/pathway · error · ValueError

The MQTT topic to read from must not be empty.

Error message

The MQTT topic to read from must not be empty.

What it means

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.

Source

Thrown at python/pathway/io/mqtt/__init__.py:181

    >>> readings = pw.io.mqtt.read(
    ...     "mqtt://localhost:1883/?client_id=sensors-reader",
    ...     "sensors/temperature",
    ...     format="json",
    ...     schema=SensorReading,
    ... )
    >>> deduplicated = readings.groupby(readings.event_id).reduce(
    ...     readings.event_id,
    ...     temperature=pw.reducers.any(readings.temperature),
    ... )

    The duplicates are exact copies of one message, so it does not matter which of
    them ``pw.reducers.any`` picks. Which field to use as the identifier depends on
    your data: a device-generated event id, a ``(device_id, timestamp)`` pair passed
    to ``groupby`` as two columns, or any other combination that uniquely identifies
    a message at the source.
    """
    if topic == "":
        raise ValueError("The MQTT topic to read from must not be empty.")

    data_storage = api.DataStorage(
        storage_type="mqtt",
        path=uri,
        topic=topic,
        mode=api.ConnectorMode.STREAMING,
        mqtt_settings=api.MqttSettings(
            qos=qos,
            retain=False,  # unused by reader
        ),
    )
    schema, data_format = construct_schema_and_data_format(
        "binary" if format == "raw" else format,
        schema=schema,
        csv_settings=None,
        json_field_paths=json_field_paths,
    )
    data_source_options = datasource.DataSourceOptions(

View on GitHub (pinned to fa2f74a464)

Solutions

1. Provide a concrete topic or a valid wildcard filter, e.g. topic='sensors/+/temperature'
2. Validate configuration at startup: assert topic, 'MQTT topic must be configured'
3. Fail fast on missing env vars instead of defaulting to empty string

Example fix

# before
topic = os.getenv("MQTT_TOPIC", "")
pw.io.mqtt.read("mqtt://localhost:1883", topic=topic)

# after
topic = os.environ["MQTT_TOPIC"]  # KeyError points at the real misconfiguration
pw.io.mqtt.read("mqtt://localhost:1883", topic=topic)

if not topic:
    raise ValueError("MQTT read topic must be a non-empty topic filter")
assert " " not in topic  # MQTT forbids spaces in topics

def is_valid_mqtt_filter(topic: str) -> bool:
    return isinstance(topic, str) and len(topic) > 0 and " " not in topic and topic.startswith("$SYS") is False or topic.startswith("$SYS/")

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• The MQTT topic to publish to must not be empty.
• The MQTT topic to publish to must not contain the wildcard c
• Failed to install dependencies
• Column {api.TIME_PSEUDOCOLUMN} cannot contain negative times
• schema does not match given dataframe

AI-assisted analysis of pathwaycom/pathway@fa2f74a464 (2026-08-15). Data as JSON: /api/errors/10d3044b77c2386b. Report an issue: GitHub.