Add portable Mqtt source and sink transforms

[ahmedabu98]

Fixes #21060

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[ahmedabu98]

CC: @chamikaramj

You can test this out locally with the following code. @gdiazdelrio can you check this PR out and give it a shot? Let me know if something doesn't work

import apache_beam as beam
from apache_beam.transforms.external_transform_provider import ExternalTransformProvider
from apache_beam.transforms.external import BeamJarExpansionService

provider = ExternalTransformProvider(BeamJarExpansionService("sdks:java:io:expansion-service:shadowJar"))
MqttRead = provider.get_urn("beam:schematransform:org.apache.beam:mqtt_read:v1")
MqttWrite = provider.get_urn("beam:schematransform:org.apache.beam:mqtt_write:v1")

with beam.Pipeline() as p:
  connection_configuration = {
    "server_uri": "tcp://localhost:58494",
    "topic": "WRITE_TOPIC",
    "client_id": "READ_PIPELINE"
  }

  # read
  p | MqttRead(connection_configuration, max_read_time_seconds=10) | beam.Map(print)

  # write
  # p | beam.Create([beam.Row(bytes=bytes([1, 2, 3, 4, 5]))]) | MqttWrite(
  #       connection_configuration=connection_configuration)

[Abacn]

Hi, what is the status of this PR? @ahmedabu98 @chamikaramj

[ahmedabu98]

@twosom perhaps you'd be interested in trying this out? I can adjust the PR to include your recently added feature to read with metadata (#32668)

[twosom]

@twosom perhaps you'd be interested in trying this out? I can adjust the PR to include your recently added feature to read with metadata (#32668)

@ahmedabu98
Thank you for the opportunity. May I give it a try?

I’ve only worked with the Java SDK so far, so this will be my first time dealing with portable development. It might take me some time, but I’ll do my best to work through it.

If I have any questions along the way, would it be alright to leave a comment here on the PR?

[ahmedabu98]

Thank you for the opportunity. May I give it a try?

Of course! You'd be doing me a favor :) Let me know how I can help

To start, you can git checkout this PR and first run ./gradlew sdks:java:io:expansion-service:build to build the necessary Java jar, then run the python code snippet I pasted above.

[twosom]

Thank you for the opportunity. May I give it a try?

Of course! You'd be doing me a favor :) Let me know how I can help

To start, you can git checkout this PR and first run ./gradlew sdks:java:io:expansion-service:build to build the necessary Java jar, then run the python code snippet I pasted above.

@ahmedabu98
Thanks!

I'll add feature for

• [Feature Request]: Add Topic Information to MqttIO Return Value #32195
• Add support for dynamic destinations when writing to MQTT #19376

and also read and write

Should I start by creating my own branch and cherry-picking your commit?