This is an automated email from the ASF dual-hosted git repository. Amar3tto pushed a commit to branch jmsio-yaml in repository https://gitbox.apache.org/repos/asf/beam.git
commit 9b5c5308bd17fdfee13779ca048839ff5c69a45a Author: Vitaly Terentyev <[email protected]> AuthorDate: Wed Aug 19 14:08:04 2026 +0400 Add JmsIO YAML test --- sdks/python/apache_beam/yaml/integration_tests.py | 23 +++++++++++++ sdks/python/apache_beam/yaml/standard_io.yaml | 28 ++++++++++++++++ sdks/python/apache_beam/yaml/tests/jms.yaml | 39 +++++++++++++++++++++++ 3 files changed, 90 insertions(+) diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index c6d73df76e3..880899efd8a 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -697,6 +697,29 @@ def temp_iceberg_table_with_pk(table_data): shutil.rmtree(temp_dir, ignore_errors=True) [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() + + @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 58080cff405..38bb624c59d 100644 --- a/sdks/python/apache_beam/yaml/standard_io.yaml +++ b/sdks/python/apache_beam/yaml/standard_io.yaml @@ -183,6 +183,34 @@ 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' + # Debezium - type: renaming transforms: 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..fd7eff8ac64 --- /dev/null +++ b/sdks/python/apache_beam/yaml/tests/jms.yaml @@ -0,0 +1,39 @@ +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
