Configuration question about mqtt -> kafka -> mqtt for topics in zilla

[jfallows]

Hello, I was at Current 2023 and saw the talk about how you're doing MQTT -> Kafka and talked with some people at the booth.

I've finally gotten around to trying to use the product and had some questions about how it might be configured for my use case.

I tried to use your demo and some of the configuration with my modifications and it doesn't quite work the way I'd hoped.

  routes:
    - when:
      - subscribe:
        - topic: devices/outgoing/+
    with:
      messages: mqtt.outgoing
    exit: north_kafka_cache_client
    when:
      - publish:
        - topic: devices/incoming/{id}
    with:
      messages: incoming_messages
    exit: north_kafka_cache_client
  exit: north_kafka_cache_client

I'd like to be able to take the incoming mqtt topic, devices/incoming/{id} and publish to the incoming_messages topic, but using the {id} bit as the key, not the full topic name.

Similarly, on the outgoing events, when reading from the mqtt.outgoing kafka topic, I'd like to use the message KEY and use it to create the outgoing topic name devices/outgoing/KEY

Not sure if that's possible since I couldn't find anything in the documentation, so I figured I'd ask here. Thank you!.

[jfallows]

The message key in Kafka is currently required to be the full mqtt topic name so we can guarantee mqtt message ordering per mqtt topic.

However this is an interesting enhancement as there may be cases where a parameter extracted from the topic name like {id} is sufficient to make the ordering guarantee.

An alternative approach is topic aliasing, which may also meet the requirement.

It depends on whether the Kafka message key actually needs to be just the {id} in Kafka topic or if the only purpose is to alias the mqtt topics.

[jgeiger-trane]

Thank you so much for posting this.

In my case the {{id}} is unique for the MQTT and Kafka topics and in Kafka that will ensure that the ordering is preserved. I'm trying to do a demo/proof of concept to see if we can use MQTT in place of a different solution, so I do require that the Kafka keys are just the ID part.

I do also need the reverse, where the incoming messages to be send OUT to the device over MQTT will need to read from a topic where the key is the ID and then prefix that value with the rest of the MQTT topic so it goes out to the correct device.

I've build a version of that functionality in Go as an MQTT/Kafka bi-directional bridge, but will be issues with load balancing and consumer groups which I think you've mostly solved already based on our discussion at Current.

I will take a look at aliases to see if it can help. I would still say that being able to specify the keys for outgoing and incoming Kafka topics would be a nice addition.

[jfallows]

@jgeiger-trane would syntax such as the following work for you?

  routes:
    - when:
      - subscribe:
        - topic: devices/outgoing/{id}+
    with:
      messages:
        topic: mqtt.outgoing
        key: ${params.id}
    exit: north_kafka_cache_client
    when:
      - publish:
        - topic: devices/incoming/{id}
    with:
      messages:
        topic: incoming_messages
        key: ${params.id}
    exit: north_kafka_cache_client
  exit: north_kafka_cache_client

[jgeiger-trane]

That looks like it would work for my use case. It'll be interesting to test it out.

Thank you!