You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by vt...@apache.org on 2026/08/19 10:08:19 UTC
(beam) branch jmsio-yaml created (now 9b5c5308bd1)
This is an automated email from the ASF dual-hosted git repository.
Amar3tto pushed a change to branch jmsio-yaml
in repository https://gitbox.apache.org/repos/asf/beam.git
at 9b5c5308bd1 Add JmsIO YAML test
This branch includes the following new commits:
new 9b5c5308bd1 Add JmsIO YAML test
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(beam) 01/01: Add JmsIO YAML test
Posted by vt...@apache.org.
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 <vi...@akvelon.com>
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)
+@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_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