JmsIO yaml

sdks/python/apache_beam/yaml/integration_tests.py

@@ -697,6 +697,55 @@ def temp_iceberg_table_with_pk(table_data):
  shutil.rmtree(temp_dir, ignore_errors=True)
 
 
+@contextlib.contextmanager
+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()
+
+
+@contextlib.contextmanager
+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.

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:

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

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

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')

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 {