This is an automated email from the ASF dual-hosted git repository.
damccorm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 88b3ee7b488 JmsIO yaml (#39818)
88b3ee7b488 is described below
commit 88b3ee7b488aa8a3e85e8dd04b64349384818d16
Author: Vitaly Terentyev <[email protected]>
AuthorDate: Fri Aug 21 18:46:38 2026 +0400
JmsIO yaml (#39818)
* Add JmsIO YAML test
* Fix depends
* Add IBM MQ
---
sdks/python/apache_beam/yaml/integration_tests.py | 49 ++++++++++++++++++
sdks/python/apache_beam/yaml/standard_io.yaml | 58 +++++++++++++++++++++
sdks/python/apache_beam/yaml/tests/ibm_mq.yaml | 62 +++++++++++++++++++++++
sdks/python/apache_beam/yaml/tests/jms.yaml | 56 ++++++++++++++++++++
sdks/python/apache_beam/yaml/yaml_provider.py | 6 ++-
sdks/python/build.gradle | 1 +
6 files changed, 230 insertions(+), 2 deletions(-)
diff --git a/sdks/python/apache_beam/yaml/integration_tests.py
b/sdks/python/apache_beam/yaml/integration_tests.py
index c6e33f8c6de..f4329205855 100644
--- a/sdks/python/apache_beam/yaml/integration_tests.py
+++ b/sdks/python/apache_beam/yaml/integration_tests.py
@@ -804,6 +804,55 @@ def temp_kinesis_localstack():
_LOGGER.info('LocalStack Kinesis fixture stopped.')
[email protected]
+def temp_jms_activemq_server():
+ """Context manager to provide a temporary ActiveMQ broker for JMS tests."""
+
+ broker =
DockerContainer('apache/activemq-classic:5.18.3').with_exposed_ports(
+ 61616)
+
+ try:
+ broker.start()
+ wait_for_logs(broker, '.*ActiveMQ .* started.*', timeout=30)
+
+ host = broker.get_container_host_ip()
+ port = broker.get_exposed_port(61616)
+
+ yield {
+ 'SERVER_URI': f'tcp://{host}:{port}',
+ 'CONNECTION_FACTORY_CLASS_NAME':
'org.apache.activemq.ActiveMQConnectionFactory',
+ }
+ finally:
+ broker.stop()
+
+
[email protected]
+def temp_ibm_mq_server():
+ container = (
+ DockerContainer('icr.io/ibm-messaging/mq:9.3.0.25-r1').with_env(
+ 'LICENSE', 'accept').with_env('MQ_QMGR_NAME', 'QM1').with_env(
+ 'MQ_APP_PASSWORD', 'admin123').with_exposed_ports(1414))
+
+ try:
+ container.start()
+ wait_for_logs(container, '.*(MQQMNAME|Started queue manager).*',
timeout=45)
+
+ host = container.get_container_host_ip()
+ port = container.get_exposed_port(1414)
+
+ yield {
+ 'SERVER_URI':
f'tcp://{host}:{port}?channel=DEV.APP.SVRCONN&queueManager=QM1',
+ 'CONNECTION_FACTORY_CLASS_NAME': 'com.ibm.mq.jms.MQConnectionFactory',
+ 'USERNAME': 'app',
+ 'PASSWORD': 'admin123',
+ 'SOURCE_QUEUE': 'DEV.QUEUE.1',
+ 'SINK_QUEUE': 'DEV.QUEUE.2',
+ }
+
+ finally:
+ container.stop()
+
+
@contextlib.contextmanager
def temp_kafka_server():
"""Context manager to provide a temporary Kafka server for testing.
diff --git a/sdks/python/apache_beam/yaml/standard_io.yaml
b/sdks/python/apache_beam/yaml/standard_io.yaml
index b14db7bbec1..81fd62b0501 100644
--- a/sdks/python/apache_beam/yaml/standard_io.yaml
+++ b/sdks/python/apache_beam/yaml/standard_io.yaml
@@ -183,6 +183,64 @@
config:
gradle_target: 'sdks:java:extensions:schemaio-expansion-service:shadowJar'
+# JMS
+- type: renaming
+ transforms:
+ 'ReadFromJms': 'ReadFromJms'
+ 'WriteToJms': 'WriteToJms'
+ config:
+ mappings:
+ 'ReadFromJms':
+ connection_configuration: 'connection_configuration'
+ queue: 'queue'
+ topic: 'topic'
+ max_num_records: 'max_num_records'
+ max_read_time_seconds: 'max_read_time_seconds'
+ close_timeout_seconds: 'close_timeout_seconds'
+ acknowledge_mode: 'acknowledge_mode'
+ individual_acknowledge_mode_code: 'individual_acknowledge_mode_code'
+ 'WriteToJms':
+ connection_configuration: 'connection_configuration'
+ queue: 'queue'
+ topic: 'topic'
+ underlying_provider:
+ type: beamJar
+ transforms:
+ 'ReadFromJms': 'beam:schematransform:org.apache.beam:jms_read:v1'
+ 'WriteToJms': 'beam:schematransform:org.apache.beam:jms_write:v1'
+ config:
+ gradle_target: 'sdks:java:io:messaging-expansion-service:shadowJar'
+- type: renaming
+ transforms:
+ 'ReadFromIbmMQ': 'ReadFromIbmMQ'
+ 'WriteToIbmMQ': 'WriteToIbmMQ'
+ config:
+ mappings:
+ 'ReadFromIbmMQ':
+ connection_configuration: 'connection_configuration'
+ queue: 'queue'
+ topic: 'topic'
+ max_num_records: 'max_num_records'
+ max_read_time_seconds: 'max_read_time_seconds'
+ close_timeout_seconds: 'close_timeout_seconds'
+ acknowledge_mode: 'acknowledge_mode'
+ individual_acknowledge_mode_code: 'individual_acknowledge_mode_code'
+ 'WriteToIbmMQ':
+ connection_configuration: 'connection_configuration'
+ queue: 'queue'
+ topic: 'topic'
+ underlying_provider:
+ type: beamJar
+ transforms:
+ 'ReadFromIbmMQ': 'beam:schematransform:org.apache.beam:jms_read:v1'
+ 'WriteToIbmMQ': 'beam:schematransform:org.apache.beam:jms_write:v1'
+ config:
+ gradle_target: 'sdks:java:io:messaging-expansion-service:shadowJar'
+ classpath:
+ - 'com.ibm.mq:com.ibm.mq.allclient:9.3.0.25'
+ - 'org.json:json:20251224'
+
+
# Debezium
- type: renaming
transforms:
diff --git a/sdks/python/apache_beam/yaml/tests/ibm_mq.yaml
b/sdks/python/apache_beam/yaml/tests/ibm_mq.yaml
new file mode 100644
index 00000000000..b621e293cc7
--- /dev/null
+++ b/sdks/python/apache_beam/yaml/tests/ibm_mq.yaml
@@ -0,0 +1,62 @@
+#
+# 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.
+#
+
+fixtures:
+ - name: IBM_MQ
+ type: apache_beam.yaml.integration_tests.temp_ibm_mq_server
+
+pipelines:
+ - pipeline:
+ type: chain
+ transforms:
+ - type: Create
+ config:
+ elements:
+ - payload: message-1
+ - payload: message-2
+ - payload: message-3
+
+ - type: WriteToIbmMQ
+ config:
+ connection_configuration:
+ server_uri: "{IBM_MQ[SERVER_URI]}"
+ connection_factory_class_name:
+ "{IBM_MQ[CONNECTION_FACTORY_CLASS_NAME]}"
+ username: "{IBM_MQ[USERNAME]}"
+ password: "{IBM_MQ[PASSWORD]}"
+ queue: "{IBM_MQ[SOURCE_QUEUE]}"
+
+ - pipeline:
+ type: chain
+ transforms:
+ - type: ReadFromIbmMQ
+ config:
+ connection_configuration:
+ server_uri: "{IBM_MQ[SERVER_URI]}"
+ connection_factory_class_name:
+ "{IBM_MQ[CONNECTION_FACTORY_CLASS_NAME]}"
+ username: "{IBM_MQ[USERNAME]}"
+ password: "{IBM_MQ[PASSWORD]}"
+ queue: "{IBM_MQ[SOURCE_QUEUE]}"
+ max_num_records: 3
+
+ - type: AssertEqual
+ config:
+ elements:
+ - payload: message-1
+ - payload: message-2
+ - payload: message-3
diff --git a/sdks/python/apache_beam/yaml/tests/jms.yaml
b/sdks/python/apache_beam/yaml/tests/jms.yaml
new file mode 100644
index 00000000000..60070dd2d45
--- /dev/null
+++ b/sdks/python/apache_beam/yaml/tests/jms.yaml
@@ -0,0 +1,56 @@
+#
+# 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.
+#
+
+fixtures:
+ - name: JMS
+ type: "apache_beam.yaml.integration_tests.temp_jms_activemq_server"
+
+pipelines:
+ - pipeline:
+ type: chain
+ transforms:
+ - type: Create
+ config:
+ elements:
+ - payload: message-1
+ - payload: message-2
+ - payload: message-3
+
+ - type: WriteToJms
+ config:
+ connection_configuration:
+ server_uri: "{JMS[SERVER_URI]}"
+ connection_factory_class_name:
"{JMS[CONNECTION_FACTORY_CLASS_NAME]}"
+ queue: yaml-jms-test
+
+ - pipeline:
+ type: chain
+ transforms:
+ - type: ReadFromJms
+ config:
+ connection_configuration:
+ server_uri: "{JMS[SERVER_URI]}"
+ connection_factory_class_name:
"{JMS[CONNECTION_FACTORY_CLASS_NAME]}"
+ queue: yaml-jms-test
+ max_num_records: 3
+
+ - type: AssertEqual
+ config:
+ elements:
+ - payload: message-1
+ - payload: message-2
+ - payload: message-3
diff --git a/sdks/python/apache_beam/yaml/yaml_provider.py
b/sdks/python/apache_beam/yaml/yaml_provider.py
index 324ae0c2e73..6dd7a502eba 100755
--- a/sdks/python/apache_beam/yaml/yaml_provider.py
+++ b/sdks/python/apache_beam/yaml/yaml_provider.py
@@ -339,12 +339,14 @@ def beam_jar(
managed_replacement=None,
appendix=None,
version=beam_version,
- artifact_id=None):
+ artifact_id=None,
+ classpath=None):
return ExternalJavaProvider(
urns, lambda: subprocess_server.JavaJarServer.path_to_beam_jar(
gradle_target=gradle_target, version=version, artifact_id=artifact_id
),
- managed_replacement=managed_replacement)
+ managed_replacement=managed_replacement,
+ classpath=classpath)
@ExternalProvider.register_provider_type('docker')
diff --git a/sdks/python/build.gradle b/sdks/python/build.gradle
index 1a93983da66..a15d4719ff7 100644
--- a/sdks/python/build.gradle
+++ b/sdks/python/build.gradle
@@ -152,6 +152,7 @@ tasks.register("yamlIntegrationTests") {
dependsOn ":sdks:java:io:expansion-service:build"
dependsOn ":sdks:java:io:google-cloud-platform:expansion-service:build"
dependsOn ":sdks:java:io:debezium:expansion-service:shadowJar"
+ dependsOn ":sdks:java:io:messaging-expansion-service:shadowJar"
doLast {
exec {
