[Feature Request]: Topic array support for MqttIO.Read

[SamLeurink]

What would you like to happen?

Currently MqttIO.ConnectionConfiguration only allows for a singular topic which in turn means that MqttIO.Read can only be done on one topic.

The use of topic arrays would, for me at least, greatly improve flexibility.

Issue Priority

Priority: 2 (default / most feature requests should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[twosom]

.take-issue

[twosom]

I think this feature is related to #32195

[twosom]

@SamLeurink
Hi,

I’ve added support for subscribing to multiple topics in MqttIO.read and MqttIO.readWithMetadata #32881.
Now you can configure multiple topics like this:

pipeline.apply(
    MqttIO.readWithMetadata()
        .withConnectionConfiguration(
            MqttIO.ConnectionConfiguration.create(
                "tcp://host:1883",
                "topic1",
                "topic2",
                "topic3"))
);

Let me know if you have any feedback!

[SamLeurink]

@twosom
Hi,

The PR is looking really good. Thank you for implementing this feature!
One thing I would maybe change is the parameter additionalTopics. If this is incorporated in the parameter topic
it would add the possibility to place Iterables such as lists in it's place. For example:

List<String> topicList = Stream.of("topic1", "topic2", "topic3")
                                  .collect(Collectors.toList());

pipeline.apply(
    MqttIO.readWithMetadata()
        .withConnectionConfiguration(
            MqttIO.ConnectionConfiguration.create(
                "tcp://host:1883",
                topicList))
);

It should only be a small change in the create method.

Let me know if it's possible to implement or if it's conflicting in any way!

[twosom]

@SamLeurink

Thank you for your reply.
I'll work on your requested feature in the same PR ^_^
Thank you.

[twosom]

@SamLeurink

Hello, I’ve updated it to accept parameters of type Iterable as requested. Please review when you have a chance.