[mqtt] Add messaging expansion service and wire MqttIO into Python xlang

Adds a new :sdks:java:io:messaging-expansion-service module that serves
messaging IOs (MQTT for now, with room for JMS/Solace later) instead of
adding MqttIO to the shared :sdks:java:io:expansion-service, per review
feedback from @Abacn and @chamikaramj.

Registers MqttIO's SchemaTransforms in standard_expansion_services.yaml
under the new service with kafka-style names (ReadFromMqtt / WriteToMqtt),
skipping the core SchemaTransforms it bundles transitively (those are
generated from the Java IO expansion service). Regenerates
standard_external_transforms.yaml so the generated Python wrappers are
served by the messaging expansion service, and registers the new target in
the generateExternalTransformsConfig task and the xlang wrapper-validation
list.

Adds an I/Os entry in CHANGES.md for 2.75.0 covering batch and streaming.

Author: tkaymak

CHANGES.md

@@ -66,6 +66,7 @@
 ## I/Os
 
 * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
+* MqttIO: portable `ReadFromMqtt` / `WriteToMqtt` transforms now available in Python via cross-language (`apache_beam.io`), for both batch and streaming ([#32385](https://github.com/apache/beam/issues/32385)).
 
 ## New Features / Improvements
 

sdks/java/io/messaging-expansion-service/build.gradle

@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * License); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an AS IS BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+apply plugin: 'org.apache.beam.module'
+apply plugin: 'application'
+mainClassName = "org.apache.beam.sdk.expansion.service.ExpansionService"
+
+applyJavaNature(
+  automaticModuleName: 'org.apache.beam.sdk.io.messaging.expansion.service',
+  exportJavadoc: false,
+  validateShadowJar: false,
+  shadowClosure: {},
+)
+
+shadowJar {
+  manifest {
+    attributes(["Multi-Release": true])
+  }
+  mergeServiceFiles()
+  outputs.upToDateWhen { false }
+}
+
+description = "Apache Beam :: SDKs :: Java :: IO :: Messaging Expansion Service"
+ext.summary = "Expansion service serving messaging IOs (e.g. MQTT)"
+
+dependencies {
+  implementation project(":sdks:java:expansion-service")
+  permitUnusedDeclared project(":sdks:java:expansion-service") // BEAM-11761
+  implementation project(":sdks:java:io:mqtt")
+  permitUnusedDeclared project(":sdks:java:io:mqtt") // BEAM-11761
+  runtimeOnly library.java.slf4j_jdk14
+}
+
+task runExpansionService (type: JavaExec) {
+  mainClass = "org.apache.beam.sdk.expansion.service.ExpansionService"
+  classpath = sourceSets.test.runtimeClasspath
+  args = [project.findProperty("constructionService.port") ?: "8097"]
+}

sdks/python/build.gradle

@@ -73,6 +73,7 @@ tasks.register("generateExternalTransformsConfig") {
   // Need to build all expansion services listed in sdks/standard_expansion_services.yaml
   dependsOn ":sdks:java:io:google-cloud-platform:expansion-service:build"
   dependsOn ":sdks:java:io:expansion-service:build"
+  dependsOn ":sdks:java:io:messaging-expansion-service:build"
   // Keep this in-sync with pyproject.toml
   def PyYaml = "'pyyaml>=3.12,<7.0.0'"
 

sdks/python/test-suites/xlang/build.gradle

@@ -25,6 +25,7 @@ project.evaluationDependsOn(":sdks:python")
 // relevant fields as done here, then add it to `xlangTasks`.
 def gcpExpansionPath = project.project(':sdks:java:io:google-cloud-platform:expansion-service').getPath()
 def ioExpansionPath = project.project(':sdks:java:io:expansion-service').getPath()
+def messagingExpansionPath = project.project(':sdks:java:io:messaging-expansion-service').getPath()
 // Properties that are common across runners.
 // Used to launch the expansion service, collect the right tests, and cleanup afterwards
 def gcpXlang = new CrossLanguageTask().tap {
@@ -42,7 +43,7 @@ def ioXlang = new CrossLanguageTask().tap {
 }
 
 // This list should include all expansion service targets in sdks/python/standard_expansion_services.yaml
-def servicesToGenerateFrom = [ioExpansionPath, gcpExpansionPath]
+def servicesToGenerateFrom = [ioExpansionPath, messagingExpansionPath, gcpExpansionPath]
 def xlangWrapperValidation = new CrossLanguageTask().tap {
     name = "xlangWrapperValidation"
     expansionProjectPaths = servicesToGenerateFrom

sdks/standard_expansion_services.yaml

@@ -53,6 +53,21 @@
     - 'beam:schematransform:org.apache.beam:iceberg_read:v1'
     - 'beam:schematransform:org.apache.beam:iceberg_cdc_read:v1'
 
+- gradle_target: 'sdks:java:io:messaging-expansion-service:shadowJar'
+  destinations:
+    python: 'apache_beam/io'
+  transforms:
+    'beam:schematransform:org.apache.beam:mqtt_write:v1':
+      name: 'WriteToMqtt'
+    'beam:schematransform:org.apache.beam:mqtt_read:v1':
+      name: 'ReadFromMqtt'
+  skip_transforms:
+    # Core SchemaTransforms bundled via :sdks:java:expansion-service; already
+    # generated from the Java IO expansion service above.
+    - 'beam:schematransform:org.apache.beam:generate_sequence:v1'
+    - 'beam:schematransform:org.apache.beam:tfrecord_read:v1'
+    - 'beam:schematransform:org.apache.beam:tfrecord_write:v1'
+
 # TODO(ahmedabu98): Enable this service in a future PR
 #- gradle_target: 'sdks:java:io:google-cloud-platform:expansion-service:shadowJar'
 #  destinations:

sdks/standard_external_transforms.yaml

@@ -19,7 +19,7 @@
 # configuration in /sdks/standard_expansion_services.yaml.
 # Refer to gen_xlang_wrappers.py for more info.
 #
-# Last updated on: 2026-05-06
+# Last updated on: 2026-06-11
 
 - default_service: sdks:java:io:expansion-service:shadowJar
   description: 'Outputs a PCollection of Beam Rows, each containing a single INT64
@@ -145,3 +145,58 @@
     type: str
   identifier: beam:schematransform:org.apache.beam:tfrecord_write:v1
   name: TfrecordWrite
+- default_service: sdks:java:io:messaging-expansion-service:shadowJar
+  description: 'Reads messages from an MQTT broker and outputs each payload as a single
+    `bytes` field.
+
+
+    By default the read is unbounded (streaming): it keeps consuming messages from
+    the subscribed topic until the pipeline is stopped. Setting `maxNumRecords` and/or
+    `maxReadTimeSeconds` bounds the read, producing a bounded (batch) PCollection.
+
+
+    Note: streaming reads require a runner that supports portable streaming (e.g.
+    Prism, Flink, or Dataflow). The legacy local Python DirectRunner does not execute
+    portable streaming cross-language reads.'
+  destinations:
+    python: apache_beam/io
+  fields:
+  - description: Configuration options to set up the MQTT connection.
+    name: connection_configuration
+    nullable: false
+    type: Row(client_id=typing.Optional[str], password=typing.Optional[str], server_uri=<class
+      'str'>, topic=typing.Optional[str], username=typing.Optional[str])
+  - description: The max number of records to receive. Setting this will result in
+      a bounded PCollection.
+    name: max_num_records
+    nullable: true
+    type: int64
+  - description: The maximum time for this source to read messages. Setting this will
+      result in a bounded PCollection.
+    name: max_read_time_seconds
+    nullable: true
+    type: int64
+  identifier: beam:schematransform:org.apache.beam:mqtt_read:v1
+  name: ReadFromMqtt
+- default_service: sdks:java:io:messaging-expansion-service:shadowJar
+  description: 'Publishes messages to an MQTT broker. Expects an input PCollection
+    of rows with a single `bytes` field, each of which is published as one MQTT message.
+
+
+    Works with both bounded (batch) and unbounded (streaming) input PCollections.'
+  destinations:
+    python: apache_beam/io
+  fields:
+  - description: Configuration options to set up the MQTT connection.
+    name: connection_configuration
+    nullable: false
+    type: Row(client_id=typing.Optional[str], password=typing.Optional[str], server_uri=<class
+      'str'>, topic=typing.Optional[str], username=typing.Optional[str])
+  - description: Whether or not the publish message should be retained by the messaging
+      engine. When a subscriber connects, it gets the latest retained message. Defaults
+      to `False`, which will clear the retained message from the server.
+    name: retained
+    nullable: true
+    type: boolean
+  identifier: beam:schematransform:org.apache.beam:mqtt_write:v1
+  name: WriteToMqtt

settings.gradle.kts

@@ -236,6 +236,7 @@ include(":sdks:java:io:elasticsearch-tests:elasticsearch-tests-8")
 include(":sdks:java:io:elasticsearch-tests:elasticsearch-tests-9")
 include(":sdks:java:io:elasticsearch-tests:elasticsearch-tests-common")
 include(":sdks:java:io:expansion-service")
+include(":sdks:java:io:messaging-expansion-service")
 include(":sdks:java:io:file-based-io-tests")
 include(":sdks:java:io:bigquery-io-perf-tests")
 include(":sdks:java:io:cdap")