This is an automated email from the ASF dual-hosted git repository.
Abacn 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 2ee432b2b61 Support JmsIO SchemaTransform and cross-lang (#39437)
2ee432b2b61 is described below
commit 2ee432b2b6101ef63b3b8d10aa7b9502e5fc36be
Author: Yi Hu <[email protected]>
AuthorDate: Tue Jul 28 16:23:10 2026 -0400
Support JmsIO SchemaTransform and cross-lang (#39437)
* [Gemini] Support JmsIO SchemaTransform and cross-lang
* Adjust timing; clean up suppresswarnings
---
...m_PostCommit_Python_Xlang_Messaging_Direct.json | 2 +-
.../prism/internal/engine/elementmanager.go | 1 +
.../java/org/apache/beam/sdk/io/jms/JmsIO.java | 145 ++++++++++++
.../sdk/io/jms/JmsReadSchemaTransformProvider.java | 228 ++++++++++++++++++
.../io/jms/JmsWriteSchemaTransformProvider.java | 192 +++++++++++++++
.../sdk/io/jms/JmsSchemaTransformProviderTest.java | 263 +++++++++++++++++++++
.../io/messaging-expansion-service/build.gradle | 3 +
.../apache_beam/io/external/xlang_jmsio_it_test.py | 235 ++++++++++++++++++
.../io/external/xlang_mqttio_it_test.py | 66 +++---
sdks/python/test-suites/direct/common.gradle | 21 +-
sdks/standard_expansion_services.yaml | 4 +
sdks/standard_external_transforms.yaml | 103 +++++++-
12 files changed, 1220 insertions(+), 43 deletions(-)
diff --git
a/.github/trigger_files/beam_PostCommit_Python_Xlang_Messaging_Direct.json
b/.github/trigger_files/beam_PostCommit_Python_Xlang_Messaging_Direct.json
index e0266d62f2e..f1ba03a243e 100644
--- a/.github/trigger_files/beam_PostCommit_Python_Xlang_Messaging_Direct.json
+++ b/.github/trigger_files/beam_PostCommit_Python_Xlang_Messaging_Direct.json
@@ -1,4 +1,4 @@
{
"comment": "Modify this file in a trivial way to cause this test suite to
run",
- "modification": 4
+ "modification": 5
}
diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go
b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go
index 3949d1af324..2b502c679db 100644
--- a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go
+++ b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go
@@ -922,6 +922,7 @@ func (em *ElementManager) PersistBundle(rb RunBundle,
col2Coders map[string]PCol
// ProcessingTime timers need to be scheduled into the processing time
based queue.
newHolds, ptRefreshes := em.triageTimers(d, inputInfo, stage)
+ // TODO(https://github.com/apache/beam/issues/39446)
// Return unprocessed to this stage's pending
// TODO sort out pending element watermark holds for process
continuation residuals.
unprocessedElements := reElementResiduals(residuals.Data, inputInfo, rb)
diff --git
a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java
b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java
index 8926d584a13..94024877320 100644
--- a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java
+++ b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java
@@ -56,6 +56,9 @@ import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.metrics.Metrics;
import org.apache.beam.sdk.options.ExecutorOptions;
import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
+import org.apache.beam.sdk.schemas.annotations.SchemaFieldDescription;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
@@ -73,6 +76,7 @@ import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.TupleTagList;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings;
import org.checkerframework.checker.initialization.qual.Initialized;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.joda.time.Duration;
@@ -208,6 +212,117 @@ public class JmsIO {
return new AutoValue_JmsIO_Write.Builder<EventT>().build();
}
+ /** A POJO describing a JMS connection. */
+ @DefaultSchema(AutoValueSchema.class)
+ @AutoValue
+ public abstract static class ConnectionConfiguration implements Serializable
{
+ public static Builder builder() {
+ return new AutoValue_JmsIO_ConnectionConfiguration.Builder();
+ }
+
+ public static ConnectionConfiguration create(
+ String serverUri, @Nullable String connectionFactoryClassName) {
+ checkArgument(serverUri != null, "serverUri can not be null");
+ return builder()
+ .setServerUri(serverUri)
+ .setConnectionFactoryClassName(connectionFactoryClassName)
+ .build();
+ }
+
+ public static ConnectionConfiguration create(String serverUri) {
+ return create(serverUri, null);
+ }
+
+ @SchemaFieldDescription("The JMS broker URI.")
+ public abstract String getServerUri();
+
+ @SchemaFieldDescription("The JMS ConnectionFactory class name.")
+ public abstract @Nullable String getConnectionFactoryClassName();
+
+ @SchemaFieldDescription("The username to connect to the JMS broker.")
+ public abstract @Nullable String getUsername();
+
+ @SchemaFieldDescription("The password to connect to the JMS broker.")
+ public abstract @Nullable String getPassword();
+
+ public ConnectionConfiguration withUsername(String username) {
+ return toBuilder().setUsername(username).build();
+ }
+
+ public ConnectionConfiguration withPassword(String password) {
+ return toBuilder().setPassword(password).build();
+ }
+
+ public ConnectionConfiguration withConnectionFactoryClassName(
+ String connectionFactoryClassName) {
+ return
toBuilder().setConnectionFactoryClassName(connectionFactoryClassName).build();
+ }
+
+ abstract Builder toBuilder();
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setServerUri(String serverUri);
+
+ public abstract Builder setConnectionFactoryClassName(
+ @Nullable String connectionFactoryClassName);
+
+ public abstract Builder setUsername(@Nullable String username);
+
+ public abstract Builder setPassword(@Nullable String password);
+
+ public abstract ConnectionConfiguration build();
+ }
+
+ public ConnectionFactory createConnectionFactory() {
+ String className = getConnectionFactoryClassName();
+ // Default to ActiveMQ
+ if (Strings.isNullOrEmpty(className)) {
+ className = "org.apache.activemq.ActiveMQConnectionFactory";
+ }
+ try {
+ Class<?> clazz = Class.forName(className);
+ String uri = getServerUri();
+ String username = getUsername();
+ String password = getPassword();
+
+ if (username != null && password != null) {
+ try {
+ return (ConnectionFactory)
+ clazz
+ .getConstructor(String.class, String.class, String.class)
+ .newInstance(username, password, uri);
+ } catch (NoSuchMethodException e) {
+ // fall through to 1-arg constructor + setters
+ }
+ }
+ try {
+ ConnectionFactory cf =
+ (ConnectionFactory)
clazz.getConstructor(String.class).newInstance(uri);
+ if (username != null && password != null) {
+ try {
+ clazz.getMethod("setUserName", String.class).invoke(cf,
username);
+ clazz.getMethod("setPassword", String.class).invoke(cf,
password);
+ } catch (NoSuchMethodException e) {
+ try {
+ clazz.getMethod("setUsername", String.class).invoke(cf,
username);
+ clazz.getMethod("setPassword", String.class).invoke(cf,
password);
+ } catch (NoSuchMethodException e2) {
+ // ignore if setters not found
+ }
+ }
+ }
+ return cf;
+ } catch (NoSuchMethodException e) {
+ return (ConnectionFactory) clazz.getConstructor().newInstance();
+ }
+ } catch (Exception e) {
+ throw new IllegalArgumentException(
+ "Unable to instantiate JMS ConnectionFactory of class " +
className, e);
+ }
+ }
+ }
+
public interface ConnectionFactoryContainer<T extends
ConnectionFactoryContainer<T>> {
T withConnectionFactory(ConnectionFactory connectionFactory);
@@ -367,6 +482,21 @@ public class JmsIO {
return
builder().setConnectionFactoryProviderFn(connectionFactoryProviderFn).build();
}
+ public Read<T> withConnectionConfiguration(ConnectionConfiguration
configuration) {
+ checkArgument(configuration != null, "configuration can not be null");
+ Read<T> read =
+ this.withConnectionFactoryProviderFn(
+ (SerializableFunction<Void, ? extends ConnectionFactory>)
+ __ -> configuration.createConnectionFactory());
+ if (configuration.getUsername() != null) {
+ read = read.withUsername(configuration.getUsername());
+ }
+ if (configuration.getPassword() != null) {
+ read = read.withPassword(configuration.getPassword());
+ }
+ return read;
+ }
+
/**
* Specify the JMS queue destination name where to read messages from. The
{@link JmsIO.Read}
* acts as a consumer on the queue.
@@ -1106,6 +1236,21 @@ public class JmsIO {
return
builder().setConnectionFactoryProviderFn(connectionFactoryProviderFn).build();
}
+ public Write<EventT> withConnectionConfiguration(ConnectionConfiguration
configuration) {
+ checkArgument(configuration != null, "configuration can not be null");
+ Write<EventT> write =
+ this.withConnectionFactoryProviderFn(
+ (SerializableFunction<Void, ? extends ConnectionFactory>)
+ __ -> configuration.createConnectionFactory());
+ if (configuration.getUsername() != null) {
+ write = write.withUsername(configuration.getUsername());
+ }
+ if (configuration.getPassword() != null) {
+ write = write.withPassword(configuration.getPassword());
+ }
+ return write;
+ }
+
/**
* Specify the JMS queue destination name where to send messages to. The
{@link JmsIO.Write}
* acts as a producer on the queue.
diff --git
a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsReadSchemaTransformProvider.java
b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsReadSchemaTransformProvider.java
new file mode 100644
index 00000000000..79643da3327
--- /dev/null
+++
b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsReadSchemaTransformProvider.java
@@ -0,0 +1,228 @@
+/*
+ * 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.
+ */
+package org.apache.beam.sdk.io.jms;
+
+import static org.apache.beam.sdk.io.jms.JmsIO.ConnectionConfiguration;
+import static
org.apache.beam.sdk.io.jms.JmsReadSchemaTransformProvider.ReadConfiguration;
+
+import com.google.auto.service.AutoService;
+import com.google.auto.value.AutoValue;
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
+import org.apache.beam.sdk.schemas.annotations.SchemaFieldDescription;
+import org.apache.beam.sdk.schemas.transforms.SchemaTransform;
+import org.apache.beam.sdk.schemas.transforms.SchemaTransformProvider;
+import org.apache.beam.sdk.schemas.transforms.TypedSchemaTransformProvider;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionRowTuple;
+import org.apache.beam.sdk.values.Row;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
+import org.joda.time.Duration;
+
+@AutoService(SchemaTransformProvider.class)
+public class JmsReadSchemaTransformProvider
+ extends TypedSchemaTransformProvider<ReadConfiguration> {
+ @DefaultSchema(AutoValueSchema.class)
+ @AutoValue
+ public abstract static class ReadConfiguration implements Serializable {
+ public static Builder builder() {
+ return new
AutoValue_JmsReadSchemaTransformProvider_ReadConfiguration.Builder();
+ }
+
+ @SchemaFieldDescription(
+ "Configuration options to set up the JMS connection.\nNote: if "
+ + "connection factory class name is set, spin up a persistent
expansion service with "
+ + "the provider client JAR on the classpath:\n"
+ + "java -cp <expansion-service-jar>:<provider-client-jar>\n"
+ + "org.apache.beam.sdk.expansion.service.ExpansionService <port>\n"
+ + "and pass expansion_service='localhost:<port>' to the transform.
Currently, only "
+ + "ActiveMQ (org.apache.activemq.ActiveMQConnectionFactory) is
embedded into the expansion service")
+ public abstract ConnectionConfiguration getConnectionConfiguration();
+
+ @SchemaFieldDescription(
+ "The JMS queue to read from. Exclusively one of queue or topic must be
specified.")
+ @Nullable
+ public abstract String getQueue();
+
+ @SchemaFieldDescription(
+ "The JMS topic to read from. Exclusively one of queue or topic must be
specified.")
+ @Nullable
+ public abstract String getTopic();
+
+ @SchemaFieldDescription(
+ "The max number of records to receive. Setting this will result in a
bounded PCollection.")
+ @Nullable
+ public abstract Long getMaxNumRecords();
+
+ @SchemaFieldDescription(
+ "The maximum time for this source to read messages. Setting this will
result in a bounded PCollection.")
+ @Nullable
+ public abstract Long getMaxReadTimeSeconds();
+
+ @SchemaFieldDescription("Close timeout for the JMS connection in seconds.")
+ @Nullable
+ public abstract Long getCloseTimeoutSeconds();
+
+ @SchemaFieldDescription(
+ "The JMS acknowledge mode: CLIENT_ACKNOWLEDGE,
CLIENT_ACKNOWLEDGE_UNSAFE, or INDIVIDUAL_ACKNOWLEDGE.")
+ @Nullable
+ public abstract String getAcknowledgeMode();
+
+ @SchemaFieldDescription(
+ "The proprietary integer code for individual message acknowledgment
when using INDIVIDUAL_ACKNOWLEDGE.")
+ @Nullable
+ public abstract Integer getIndividualAcknowledgeModeCode();
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setConnectionConfiguration(
+ ConnectionConfiguration connectionConfiguration);
+
+ public abstract Builder setQueue(String queue);
+
+ public abstract Builder setTopic(String topic);
+
+ public abstract Builder setMaxNumRecords(Long maxNumRecords);
+
+ public abstract Builder setMaxReadTimeSeconds(Long maxReadTimeSeconds);
+
+ public abstract Builder setCloseTimeoutSeconds(Long closeTimeoutSeconds);
+
+ public abstract Builder setAcknowledgeMode(String acknowledgeMode);
+
+ public abstract Builder setIndividualAcknowledgeModeCode(
+ Integer individualAcknowledgeModeCode);
+
+ public abstract ReadConfiguration build();
+ }
+ }
+
+ @Override
+ public String identifier() {
+ return "beam:schematransform:org.apache.beam:jms_read:v1";
+ }
+
+ @Override
+ public String description() {
+ return "Reads messages from a JMS broker and outputs each message as a
single `payload` "
+ + "(string) field.\n"
+ + "\n"
+ + "By default the read is unbounded (streaming): it keeps consuming
messages from the "
+ + "specified queue or topic until the pipeline is stopped. Setting
`maxNumRecords` and/or "
+ + "`maxReadTimeSeconds` bounds the read, producing a bounded (batch)
PCollection.";
+ }
+
+ @Override
+ public List<String> outputCollectionNames() {
+ return Collections.singletonList("output");
+ }
+
+ @Override
+ protected SchemaTransform from(ReadConfiguration configuration) {
+ return new JmsReadSchemaTransform(configuration);
+ }
+
+ private static class JmsReadSchemaTransform extends SchemaTransform {
+ private final ReadConfiguration config;
+
+ JmsReadSchemaTransform(ReadConfiguration configuration) {
+ this.config = configuration;
+ }
+
+ @Override
+ public PCollectionRowTuple expand(PCollectionRowTuple input) {
+ Preconditions.checkState(
+ input.getAll().isEmpty(),
+ "Expected zero input PCollections for this source, but found: %s",
+ input.getAll().keySet());
+
+ Preconditions.checkArgument(
+ (config.getQueue() != null) != (config.getTopic() != null),
+ "Exactly one of queue or topic must be specified.");
+
+ JmsIO.Read<JmsRecord> readTransform =
+
JmsIO.read().withConnectionConfiguration(config.getConnectionConfiguration());
+
+ String queue = config.getQueue();
+ if (queue != null) {
+ readTransform = readTransform.withQueue(queue);
+ }
+ String topic = config.getTopic();
+ if (topic != null) {
+ readTransform = readTransform.withTopic(topic);
+ }
+
+ Long maxRecords = config.getMaxNumRecords();
+ Long maxReadTime = config.getMaxReadTimeSeconds();
+ if (maxRecords != null) {
+ readTransform = readTransform.withMaxNumRecords(maxRecords);
+ }
+ if (maxReadTime != null) {
+ readTransform =
readTransform.withMaxReadTime(Duration.standardSeconds(maxReadTime));
+ }
+
+ Long closeTimeout = config.getCloseTimeoutSeconds();
+ if (closeTimeout != null) {
+ readTransform =
readTransform.withCloseTimeout(Duration.standardSeconds(closeTimeout));
+ }
+
+ String ackMode = config.getAcknowledgeMode();
+ if (ackMode != null) {
+ readTransform =
+
readTransform.withAcknowledgeMode(JmsIO.AcknowledgeMode.valueOf(ackMode.toUpperCase()));
+ }
+
+ Integer indAckCode = config.getIndividualAcknowledgeModeCode();
+ if (indAckCode != null) {
+ readTransform =
readTransform.withIndividualAcknowledgeModeCode(indAckCode);
+ }
+
+ Schema outputSchema = Schema.builder().addStringField("payload").build();
+
+ PCollection<Row> outputRows =
+ input
+ .getPipeline()
+ .apply(readTransform)
+ .apply(
+ "Wrap in Beam Rows",
+ ParDo.of(
+ new DoFn<JmsRecord, Row>() {
+ @ProcessElement
+ public void processElement(
+ @Element JmsRecord record, OutputReceiver<Row>
outputReceiver) {
+ String payload = record.getPayload();
+ if (payload == null) {
+ payload = "";
+ }
+ outputReceiver.output(
+
Row.withSchema(outputSchema).addValue(payload).build());
+ }
+ }))
+ .setRowSchema(outputSchema);
+
+ return PCollectionRowTuple.of("output", outputRows);
+ }
+ }
+}
diff --git
a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsWriteSchemaTransformProvider.java
b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsWriteSchemaTransformProvider.java
new file mode 100644
index 00000000000..5db7271b4b9
--- /dev/null
+++
b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsWriteSchemaTransformProvider.java
@@ -0,0 +1,192 @@
+/*
+ * 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.
+ */
+package org.apache.beam.sdk.io.jms;
+
+import static org.apache.beam.sdk.io.jms.JmsIO.ConnectionConfiguration;
+import static
org.apache.beam.sdk.io.jms.JmsWriteSchemaTransformProvider.WriteConfiguration;
+
+import com.google.auto.service.AutoService;
+import com.google.auto.value.AutoValue;
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
+import org.apache.beam.sdk.schemas.annotations.SchemaFieldDescription;
+import org.apache.beam.sdk.schemas.transforms.SchemaTransform;
+import org.apache.beam.sdk.schemas.transforms.SchemaTransformProvider;
+import org.apache.beam.sdk.schemas.transforms.TypedSchemaTransformProvider;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionRowTuple;
+import org.apache.beam.sdk.values.Row;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
+
+@AutoService(SchemaTransformProvider.class)
+public class JmsWriteSchemaTransformProvider
+ extends TypedSchemaTransformProvider<WriteConfiguration> {
+ @DefaultSchema(AutoValueSchema.class)
+ @AutoValue
+ public abstract static class WriteConfiguration implements Serializable {
+ public static Builder builder() {
+ return new
AutoValue_JmsWriteSchemaTransformProvider_WriteConfiguration.Builder();
+ }
+
+ @SchemaFieldDescription(
+ "Configuration options to set up the JMS connection.\nNote: if "
+ + "connection factory class name is set, spin up a persistent
expansion service with "
+ + "the provider client JAR on the classpath:\n"
+ + "java -cp <expansion-service-jar>:<provider-client-jar>\n"
+ + "org.apache.beam.sdk.expansion.service.ExpansionService <port>\n"
+ + "and pass expansion_service='localhost:<port>' to the transform.
Currently, only "
+ + "ActiveMQ (org.apache.activemq.ActiveMQConnectionFactory) is
embedded into the expansion service")
+ public abstract ConnectionConfiguration getConnectionConfiguration();
+
+ @SchemaFieldDescription(
+ "The JMS queue to write to. Exclusively one of queue or topic must be
specified.")
+ @Nullable
+ public abstract String getQueue();
+
+ @SchemaFieldDescription(
+ "The JMS topic to write to. Exclusively one of queue or topic must be
specified.")
+ @Nullable
+ public abstract String getTopic();
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setConnectionConfiguration(
+ ConnectionConfiguration connectionConfiguration);
+
+ public abstract Builder setQueue(String queue);
+
+ public abstract Builder setTopic(String topic);
+
+ public abstract WriteConfiguration build();
+ }
+ }
+
+ @Override
+ public String identifier() {
+ return "beam:schematransform:org.apache.beam:jms_write:v1";
+ }
+
+ @Override
+ public String description() {
+ return "Publishes messages to a JMS broker. Expects an input PCollection
of rows with a "
+ + "`payload` (string) or `bytes` field, each of which is published as
one JMS TextMessage.\n"
+ + "\n"
+ + "Works with both bounded (batch) and unbounded (streaming) input
PCollections.";
+ }
+
+ @Override
+ public List<String> inputCollectionNames() {
+ return Collections.singletonList("input");
+ }
+
+ @Override
+ protected SchemaTransform from(WriteConfiguration configuration) {
+ return new JmsWriteSchemaTransform(configuration);
+ }
+
+ private static class JmsWriteSchemaTransform extends SchemaTransform {
+ private final WriteConfiguration config;
+
+ JmsWriteSchemaTransform(WriteConfiguration configuration) {
+ this.config = configuration;
+ }
+
+ @Override
+ public PCollectionRowTuple expand(PCollectionRowTuple input) {
+ PCollection<Row> inputRows = input.getSinglePCollection();
+
+ Preconditions.checkArgument(
+ (config.getQueue() != null) != (config.getTopic() != null),
+ "Exactly one of queue or topic must be specified.");
+
+ int fieldIndex = -1;
+ boolean isBytes = false;
+ Schema schema = inputRows.getSchema();
+ if (schema.hasField("payload")
+ &&
schema.getField("payload").getType().equals(Schema.FieldType.STRING)) {
+ fieldIndex = schema.indexOf("payload");
+ } else if (schema.hasField("bytes")
+ &&
schema.getField("bytes").getType().equals(Schema.FieldType.BYTES)) {
+ fieldIndex = schema.indexOf("bytes");
+ isBytes = true;
+ } else if (schema.getFieldCount() == 1
+ && schema.getField(0).getType().equals(Schema.FieldType.STRING)) {
+ fieldIndex = 0;
+ } else if (schema.getFieldCount() == 1
+ && schema.getField(0).getType().equals(Schema.FieldType.BYTES)) {
+ fieldIndex = 0;
+ isBytes = true;
+ } else {
+ throw new IllegalStateException(
+ String.format(
+ "Expected input Schema to have a 'payload' (STRING) or 'bytes'
(BYTES) field, or"
+ + " a single string/bytes field, but received: %s",
+ schema));
+ }
+
+ JmsIO.Write<String> writeTransform =
+ JmsIO.<String>write()
+ .withConnectionConfiguration(config.getConnectionConfiguration())
+ .withValueMapper(new TextMessageMapper());
+
+ String queue = config.getQueue();
+ if (queue != null) {
+ writeTransform = writeTransform.withQueue(queue);
+ }
+ String topic = config.getTopic();
+ if (topic != null) {
+ writeTransform = writeTransform.withTopic(topic);
+ }
+
+ final int idx = fieldIndex;
+ final boolean bytesField = isBytes;
+ inputRows
+ .apply(
+ "Extract payload",
+ ParDo.of(
+ new DoFn<Row, String>() {
+ @ProcessElement
+ public void processElement(
+ @Element Row row, OutputReceiver<String>
outputReceiver) {
+ if (bytesField) {
+ byte[] bytes =
+
org.apache.beam.sdk.util.Preconditions.checkStateNotNull(
+ row.getBytes(idx));
+ outputReceiver.output(new String(bytes,
StandardCharsets.UTF_8));
+ } else {
+ String payload =
+
org.apache.beam.sdk.util.Preconditions.checkStateNotNull(
+ row.getString(idx));
+ outputReceiver.output(payload);
+ }
+ }
+ }))
+ .apply(writeTransform);
+
+ return PCollectionRowTuple.empty(inputRows.getPipeline());
+ }
+ }
+}
diff --git
a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsSchemaTransformProviderTest.java
b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsSchemaTransformProviderTest.java
new file mode 100644
index 00000000000..b354ef94a00
--- /dev/null
+++
b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsSchemaTransformProviderTest.java
@@ -0,0 +1,263 @@
+/*
+ * 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.
+ */
+package org.apache.beam.sdk.io.jms;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.util.List;
+import java.util.ServiceLoader;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+import
org.apache.beam.sdk.io.jms.JmsReadSchemaTransformProvider.ReadConfiguration;
+import
org.apache.beam.sdk.io.jms.JmsWriteSchemaTransformProvider.WriteConfiguration;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.transforms.SchemaTransform;
+import org.apache.beam.sdk.schemas.transforms.SchemaTransformProvider;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionRowTuple;
+import org.apache.beam.sdk.values.Row;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Sets;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link JmsReadSchemaTransformProvider} and {@link
JmsWriteSchemaTransformProvider}. */
+@RunWith(JUnit4.class)
+public class JmsSchemaTransformProviderTest {
+
+ @Rule
+ public final transient TestPipeline pipeline =
+ TestPipeline.create().enableAbandonedNodeEnforcement(false);
+
+ @Test
+ public void testReadFindTransform() {
+ ServiceLoader<SchemaTransformProvider> serviceLoader =
+ ServiceLoader.load(SchemaTransformProvider.class);
+ List<SchemaTransformProvider> providers =
+ StreamSupport.stream(serviceLoader.spliterator(), false)
+ .filter(provider -> provider.getClass() ==
JmsReadSchemaTransformProvider.class)
+ .collect(Collectors.toList());
+ SchemaTransformProvider jmsProvider = providers.get(0);
+
+ assertEquals(Lists.newArrayList("output"),
jmsProvider.outputCollectionNames());
+ assertEquals(Lists.newArrayList(), jmsProvider.inputCollectionNames());
+ assertEquals("beam:schematransform:org.apache.beam:jms_read:v1",
jmsProvider.identifier());
+ assertNotNull(jmsProvider.description());
+
+ assertEquals(
+ Sets.newHashSet(
+ "connection_configuration",
+ "queue",
+ "topic",
+ "max_num_records",
+ "max_read_time_seconds",
+ "close_timeout_seconds",
+ "acknowledge_mode",
+ "individual_acknowledge_mode_code"),
+ jmsProvider.configurationSchema().getFields().stream()
+ .map(Schema.Field::getName)
+ .collect(Collectors.toSet()));
+ }
+
+ @Test
+ public void testReadBuildTransformWithQueue() {
+ ReadConfiguration readConfig =
+ ReadConfiguration.builder()
+ .setConnectionConfiguration(
+ JmsIO.ConnectionConfiguration.create("tcp://localhost:61616"))
+ .setQueue("TEST_QUEUE")
+ .setMaxNumRecords(100L)
+ .setMaxReadTimeSeconds(5L)
+ .setCloseTimeoutSeconds(10L)
+ .setAcknowledgeMode("CLIENT_ACKNOWLEDGE")
+ .setIndividualAcknowledgeModeCode(101)
+ .build();
+
+ SchemaTransform transform = new
JmsReadSchemaTransformProvider().from(readConfig);
+ PCollectionRowTuple output =
transform.expand(PCollectionRowTuple.empty(pipeline));
+
+ assertEquals(1, output.getAll().size());
+ assertTrue(output.has("output"));
+ assertEquals(
+ Schema.builder().addStringField("payload").build(),
output.get("output").getSchema());
+ }
+
+ @Test
+ public void testReadBuildTransformWithTopic() {
+ ReadConfiguration readConfig =
+ ReadConfiguration.builder()
+ .setConnectionConfiguration(
+ JmsIO.ConnectionConfiguration.create("tcp://localhost:61616"))
+ .setTopic("TEST_TOPIC")
+ .build();
+
+ SchemaTransform transform = new
JmsReadSchemaTransformProvider().from(readConfig);
+ PCollectionRowTuple output =
transform.expand(PCollectionRowTuple.empty(pipeline));
+
+ assertEquals(1, output.getAll().size());
+ assertTrue(output.has("output"));
+ assertEquals(
+ Schema.builder().addStringField("payload").build(),
output.get("output").getSchema());
+ }
+
+ @Test
+ public void testReadInvalidConfigurations() {
+ ReadConfiguration bothConfig =
+ ReadConfiguration.builder()
+ .setConnectionConfiguration(
+ JmsIO.ConnectionConfiguration.create("tcp://localhost:61616"))
+ .setQueue("TEST_QUEUE")
+ .setTopic("TEST_TOPIC")
+ .build();
+ SchemaTransform bothTransform = new
JmsReadSchemaTransformProvider().from(bothConfig);
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> bothTransform.expand(PCollectionRowTuple.empty(pipeline)));
+
+ ReadConfiguration neitherConfig =
+ ReadConfiguration.builder()
+ .setConnectionConfiguration(
+ JmsIO.ConnectionConfiguration.create("tcp://localhost:61616"))
+ .build();
+ SchemaTransform neitherTransform = new
JmsReadSchemaTransformProvider().from(neitherConfig);
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> neitherTransform.expand(PCollectionRowTuple.empty(pipeline)));
+ }
+
+ @Test
+ public void testReadWithNonEmptyInputThrows() {
+ ReadConfiguration readConfig =
+ ReadConfiguration.builder()
+ .setConnectionConfiguration(
+ JmsIO.ConnectionConfiguration.create("tcp://localhost:61616"))
+ .setQueue("TEST_QUEUE")
+ .build();
+ SchemaTransform transform = new
JmsReadSchemaTransformProvider().from(readConfig);
+
+ PCollection<Row> dummyInput =
+ pipeline.apply(
+ "CreateDummy",
Create.empty(Schema.builder().addStringField("dummy").build()));
+ assertThrows(
+ IllegalStateException.class,
+ () -> transform.expand(PCollectionRowTuple.of("input", dummyInput)));
+ }
+
+ @Test
+ public void testWriteFindTransform() {
+ ServiceLoader<SchemaTransformProvider> serviceLoader =
+ ServiceLoader.load(SchemaTransformProvider.class);
+ List<SchemaTransformProvider> providers =
+ StreamSupport.stream(serviceLoader.spliterator(), false)
+ .filter(provider -> provider.getClass() ==
JmsWriteSchemaTransformProvider.class)
+ .collect(Collectors.toList());
+ SchemaTransformProvider jmsProvider = providers.get(0);
+
+ assertEquals(Lists.newArrayList(), jmsProvider.outputCollectionNames());
+ assertEquals(Lists.newArrayList("input"),
jmsProvider.inputCollectionNames());
+ assertEquals("beam:schematransform:org.apache.beam:jms_write:v1",
jmsProvider.identifier());
+ assertNotNull(jmsProvider.description());
+
+ assertEquals(
+ Sets.newHashSet("connection_configuration", "queue", "topic"),
+ jmsProvider.configurationSchema().getFields().stream()
+ .map(Schema.Field::getName)
+ .collect(Collectors.toSet()));
+ }
+
+ @Test
+ public void testWriteBuildTransformWithQueueAndTopic() {
+ WriteConfiguration queueConfig =
+ WriteConfiguration.builder()
+ .setConnectionConfiguration(
+ JmsIO.ConnectionConfiguration.create("tcp://localhost:61616"))
+ .setQueue("TEST_QUEUE")
+ .build();
+ SchemaTransform queueTransform = new
JmsWriteSchemaTransformProvider().from(queueConfig);
+ Schema schema = Schema.builder().addStringField("payload").build();
+ PCollection<Row> inputRows = pipeline.apply("CreateQueueRows",
Create.empty(schema));
+ PCollectionRowTuple output =
queueTransform.expand(PCollectionRowTuple.of("input", inputRows));
+ assertTrue(output.getAll().isEmpty());
+
+ WriteConfiguration topicConfig =
+ WriteConfiguration.builder()
+ .setConnectionConfiguration(
+ JmsIO.ConnectionConfiguration.create("tcp://localhost:61616"))
+ .setTopic("TEST_TOPIC")
+ .build();
+ SchemaTransform topicTransform = new
JmsWriteSchemaTransformProvider().from(topicConfig);
+ Schema bytesSchema = Schema.builder().addByteArrayField("bytes").build();
+ PCollection<Row> bytesRows = pipeline.apply("CreateTopicRows",
Create.empty(bytesSchema));
+ PCollectionRowTuple topicOutput =
+ topicTransform.expand(PCollectionRowTuple.of("input", bytesRows));
+ assertTrue(topicOutput.getAll().isEmpty());
+ }
+
+ @Test
+ public void testWriteInvalidConfigurations() {
+ WriteConfiguration bothConfig =
+ WriteConfiguration.builder()
+ .setConnectionConfiguration(
+ JmsIO.ConnectionConfiguration.create("tcp://localhost:61616"))
+ .setQueue("TEST_QUEUE")
+ .setTopic("TEST_TOPIC")
+ .build();
+ SchemaTransform bothTransform = new
JmsWriteSchemaTransformProvider().from(bothConfig);
+ Schema schema = Schema.builder().addStringField("payload").build();
+ PCollection<Row> inputRows = pipeline.apply("CreateBothRows",
Create.empty(schema));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> bothTransform.expand(PCollectionRowTuple.of("input",
inputRows)));
+
+ WriteConfiguration neitherConfig =
+ WriteConfiguration.builder()
+ .setConnectionConfiguration(
+ JmsIO.ConnectionConfiguration.create("tcp://localhost:61616"))
+ .build();
+ SchemaTransform neitherTransform = new
JmsWriteSchemaTransformProvider().from(neitherConfig);
+ PCollection<Row> inputRows2 = pipeline.apply("CreateNeitherRows",
Create.empty(schema));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> neitherTransform.expand(PCollectionRowTuple.of("input",
inputRows2)));
+ }
+
+ @Test
+ public void testWriteInvalidInputSchema() {
+ WriteConfiguration config =
+ WriteConfiguration.builder()
+ .setConnectionConfiguration(
+ JmsIO.ConnectionConfiguration.create("tcp://localhost:61616"))
+ .setQueue("TEST_QUEUE")
+ .build();
+ SchemaTransform transform = new
JmsWriteSchemaTransformProvider().from(config);
+
+ Schema invalidSchema =
Schema.builder().addInt32Field("id").addStringField("name").build();
+ PCollection<Row> inputRows =
+ pipeline.apply("CreateInvalidSchemaRows", Create.empty(invalidSchema));
+ assertThrows(
+ IllegalStateException.class,
+ () -> transform.expand(PCollectionRowTuple.of("input", inputRows)));
+ }
+}
diff --git a/sdks/java/io/messaging-expansion-service/build.gradle
b/sdks/java/io/messaging-expansion-service/build.gradle
index 6cdf67b86d6..8693f4597bc 100644
--- a/sdks/java/io/messaging-expansion-service/build.gradle
+++ b/sdks/java/io/messaging-expansion-service/build.gradle
@@ -42,6 +42,9 @@ dependencies {
permitUnusedDeclared project(":sdks:java:expansion-service") // BEAM-11761
implementation project(":sdks:java:io:mqtt")
permitUnusedDeclared project(":sdks:java:io:mqtt") // BEAM-11761
+ implementation project(":sdks:java:io:jms")
+ permitUnusedDeclared project(":sdks:java:io:jms") // BEAM-11761
+ runtimeOnly library.java.activemq_client
runtimeOnly library.java.slf4j_jdk14
}
diff --git a/sdks/python/apache_beam/io/external/xlang_jmsio_it_test.py
b/sdks/python/apache_beam/io/external/xlang_jmsio_it_test.py
new file mode 100644
index 00000000000..98905107f05
--- /dev/null
+++ b/sdks/python/apache_beam/io/external/xlang_jmsio_it_test.py
@@ -0,0 +1,235 @@
+#
+# 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.
+#
+
+"""Integration tests for the cross-language JMS IO transforms
+(ReadFromJms / WriteToJms), served by the messaging expansion service.
+
+Runs against an ActiveMQ broker started once per test class via testcontainers.
+"""
+
+import logging
+import threading
+import time
+import unittest
+
+import pytest
+
+import apache_beam as beam
+from apache_beam.options.pipeline_options import PortableOptions
+from apache_beam.options.pipeline_options import StandardOptions
+from apache_beam.testing.test_pipeline import TestPipeline
+from apache_beam.typehints.row_type import RowTypeConstraint
+
+# pylint: disable=wrong-import-order, wrong-import-position, ungrouped-imports
+try:
+ from apache_beam.io import ReadFromJms
+ from apache_beam.io import WriteToJms
+except ImportError:
+ ReadFromJms = None
+ WriteToJms = None
+
+try:
+ from testcontainers.core.container import DockerContainer
+ from testcontainers.core.waiting_utils import wait_for_logs
+except ImportError:
+ DockerContainer = None
+
+_LOGGER = logging.getLogger(__name__)
+NUM_RECORDS = 100
+STRING_ROW = RowTypeConstraint.from_fields([('payload', str)])
+
+
[email protected]_messaging_java_expansion_service
[email protected](
+ DockerContainer is None, 'testcontainers package is not installed')
[email protected](
+ ReadFromJms is None or WriteToJms is None,
+ 'JMS cross-language wrappers are not generated')
[email protected](
+ TestPipeline().get_pipeline_options().view_as(StandardOptions).runner
+ is None,
+ 'Do not run this test on precommit suites.')
[email protected](
+ 'Dataflow' in (
+ TestPipeline().get_pipeline_options().view_as(StandardOptions).runner
or
+ ''),
+ 'The testcontainers broker is not reachable from Dataflow workers; '
+ 'a Dataflow variant would need a remotely hosted JMS broker.')
+class CrossLanguageJmsIOTest(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ cls.start_jms_container(retries=3)
+ host = cls.broker.get_container_host_ip()
+ port = cls.broker.get_exposed_port(61616)
+ cls.server_uri = 'tcp://%s:%s' % (host, port)
+
+ @classmethod
+ def tearDownClass(cls):
+ try:
+ cls.broker.stop()
+ except Exception:
+ logging.error('Could not stop the JMS broker container.')
+
+ @classmethod
+ def start_jms_container(cls, retries):
+ for i in range(retries):
+ try:
+ cls.broker = DockerContainer(
+ 'apache/activemq-classic:5.18.3').with_exposed_ports(61616)
+ cls.broker.start()
+ wait_for_logs(cls.broker, '.*ActiveMQ .* started.*', timeout=30)
+ break
+ except Exception as e:
+ try:
+ cls.broker.stop()
+ except Exception:
+ pass
+ if i == retries - 1:
+ logging.error('Unable to initialize the JMS broker container.')
+ raise e
+
+ def _connection_configuration(self, connection_param=None):
+ uri = self.server_uri
+ if connection_param:
+ uri += '?' + connection_param
+ return {
+ 'server_uri': uri,
+ 'connection_factory_class_name':
'org.apache.activemq.ActiveMQConnectionFactory'
+ }
+
+ def _run_streaming_test(
+ self,
+ source_queue,
+ sink_queue,
+ acknowledge_mode=None,
+ connection_param=None):
+ subscriber_result = {}
+
+ def produce(count):
+ container = self.broker.get_wrapped_container()
+ exit_code, _ = container.exec_run([
+ '/opt/apache-activemq/bin/activemq',
+ 'producer',
+ '--destination',
+ 'queue://' + source_queue,
+ '--messageCount',
+ str(count),
+ '--persistent',
+ 'false'
+ ])
+ if exit_code == 0:
+ _LOGGER.info('published %s messages', count)
+ else:
+ _LOGGER.warning('publishing message returns exit code %s', exit_code)
+
+ def publish():
+ produce(remaining_records)
+
+ stop_event = threading.Event()
+
+ def subscribe():
+ # Poll the sink queue every few seconds until NUM_RECORDS messages arrive
+ # or timeout occurs, avoiding blocking indefinitely inside activemq
consumer.
+ container = self.broker.get_wrapped_container()
+ received_messages = []
+ while len(received_messages) < NUM_RECORDS and not stop_event.is_set():
+ time.sleep(5)
+ try:
+ exit_code, output = container.exec_run([
+ '/opt/apache-activemq/bin/activemq',
+ 'browse',
+ sink_queue
+ ])
+ if exit_code == 0 and output:
+ received_messages = [
+ line.split('JMS_BODY_FIELD:JMSText = ')[-1].strip()
+ for line in output.decode('utf-8').splitlines()
+ if 'JMS_BODY_FIELD:JMSText = ' in line
+ ]
+ subscriber_result['received'] = received_messages
+ except Exception as e:
+ _LOGGER.warning('Error while browsing sink queue: %s', e)
+ break
+ _LOGGER.info('received %s messages', len(received_messages))
+
+ # TODO(https://github.com/apache/beam/issues/39446): Clean up
+ # pre-publishing Prism runner issue resolved
+ initial_records = 10
+ remaining_records = NUM_RECORDS - initial_records
+ produce(initial_records)
+
+ publisher = threading.Thread(target=publish, daemon=True)
+ subscriber = threading.Thread(target=subscribe, daemon=True)
+
+ p = TestPipeline(blocking=False)
+ p.get_pipeline_options().view_as(StandardOptions).streaming = True
+ p.not_use_test_runner_api = True
+ # Run pipeline without blocking
+ # TODO: Remove once subprocess cache leak fixed for pipeline running
+ # in LOOPBACK mode outside of with clause
+ p.__enter__()
+ try:
+ _ = (
+ p
+ | 'ReadFromJms' >> ReadFromJms(
+ connection_configuration=self._connection_configuration(
+ connection_param),
+ queue=source_queue,
+ acknowledge_mode=acknowledge_mode)
+ |
+ 'Passthrough' >> beam.Map(lambda row: beam.Row(payload=row.payload)
+ ).with_output_types(STRING_ROW)
+ | 'WriteToJms' >> WriteToJms(
+ connection_configuration=self._connection_configuration(
+ connection_param),
+ queue=sink_queue))
+ publisher.start()
+ result = p.run()
+ subscriber.start()
+ try:
+ subscriber.join(timeout=90) # 1.5 min
+ finally:
+ stop_event.set()
+ publisher.join()
+ try:
+ result.cancel()
+ except Exception: # pylint: disable=broad-except
+ _LOGGER.warning('Ignoring error while cancelling the pipeline.')
+ finally:
+ p._extra_context.__exit__(None, None, None)
+
+ received = subscriber_result.get('received', [])
+ self.assertEqual(len(received), NUM_RECORDS)
+ # there are identical records
+ self.assertEqual(len(set(received)), NUM_RECORDS - initial_records)
+
+ def test_xlang_jms_write_read_queue_ind_ack(self):
+ self._run_streaming_test(
+ source_queue='xlang-jms-ind-source',
+ sink_queue='xlang-jms-ind-sink',
+ acknowledge_mode='INDIVIDUAL_ACKNOWLEDGE')
+
+ def test_xlang_jms_write_read_queue(self):
+ self._run_streaming_test(
+ source_queue='xlang-jms-source',
+ sink_queue='xlang-jms-sink',
+ connection_param='jms.prefetchPolicy.all=0')
+
+
+if __name__ == '__main__':
+ logging.getLogger().setLevel(logging.INFO)
+ unittest.main()
diff --git a/sdks/python/apache_beam/io/external/xlang_mqttio_it_test.py
b/sdks/python/apache_beam/io/external/xlang_mqttio_it_test.py
index 889096d1f22..dad85a688ec 100644
--- a/sdks/python/apache_beam/io/external/xlang_mqttio_it_test.py
+++ b/sdks/python/apache_beam/io/external/xlang_mqttio_it_test.py
@@ -214,46 +214,40 @@ class CrossLanguageMqttIOTest(unittest.TestCase):
publisher.start()
subscriber.start()
- # MqttIO read is unbounded, so this pipeline runs in streaming mode and
- # never terminates on its own. Amend the harness-provided pipeline options
- # rather than discarding them: enable streaming, run non-blocking so the
- # observe-then-cancel logic below can execute, and target the Prism
portable
- # runner. The latter is required because SwitchingDirectRunner disables its
- # Prism delegation for pipelines containing external (cross-language)
- # transforms (see runners/direct/direct_runner.py) and falls back to the
- # BundleBasedDirectRunner, which cannot execute an unbounded read.
- # The runner is instantiated during TestPipeline construction, so it must
be
- # passed to the constructor; the remaining harness-provided options are
- # preserved and only amended (streaming + LOOPBACK environment) afterwards.
- p = TestPipeline(runner='PrismRunner', blocking=False)
+ p = TestPipeline(blocking=False)
p.get_pipeline_options().view_as(StandardOptions).streaming = True
- p.get_pipeline_options().view_as(
- PortableOptions).environment_type = 'LOOPBACK'
p.not_use_test_runner_api = True
- _ = (
- p
- | 'ReadFromMqtt' >> ReadFromMqtt(
- connection_configuration=self._connection_configuration(
- source_topic, 'xlang-mqtt-streaming-read'))
- | 'Passthrough' >> beam.Map(
- lambda row: beam.Row(bytes=row.bytes)).with_output_types(BYTES_ROW)
- | 'WriteToMqtt' >> WriteToMqtt(
- connection_configuration=self._connection_configuration(
- sink_topic, 'xlang-mqtt-streaming-write')))
- result = p.run()
+ # Run pipeline without blocking
+ # TODO: Remove once subprocess cache leak fixed for pipeline running
+ # in LOOPBACK mode outside of with clause
+ p.__enter__()
try:
- # The subscriber exits once NUM_RECORDS messages flowed through the
- # streaming pipeline (or fails the assertions below on its timeout).
- subscriber.join(timeout=200)
- finally:
- stop_publishing.set()
- publisher.join()
+ _ = (
+ p
+ | 'ReadFromMqtt' >> ReadFromMqtt(
+ connection_configuration=self._connection_configuration(
+ source_topic, 'xlang-mqtt-streaming-read'))
+ | 'Passthrough' >> beam.Map(lambda row: beam.Row(bytes=row.bytes)).
+ with_output_types(BYTES_ROW)
+ | 'WriteToMqtt' >> WriteToMqtt(
+ connection_configuration=self._connection_configuration(
+ sink_topic, 'xlang-mqtt-streaming-write')))
+ result = p.run()
try:
- result.cancel()
- except Exception: # pylint: disable=broad-except
- # The unbounded pipeline never finishes on its own; cancellation
- # after the assertion data was collected is best-effort.
- logging.warning('Ignoring error while cancelling the pipeline.')
+ # The subscriber exits once NUM_RECORDS messages flowed through the
+ # streaming pipeline (or fails the assertions below on its timeout).
+ subscriber.join(timeout=200)
+ finally:
+ stop_publishing.set()
+ publisher.join()
+ try:
+ result.cancel()
+ except Exception: # pylint: disable=broad-except
+ # The unbounded pipeline never finishes on its own; cancellation
+ # after the assertion data was collected is best-effort.
+ logging.warning('Ignoring error while cancelling the pipeline.')
+ finally:
+ p._extra_context.__exit__(None, None, None)
self.assertEqual(subscriber_result.get('exit_code'), 0)
payloads = subscriber_result.get('output', b'').split()
diff --git a/sdks/python/test-suites/direct/common.gradle
b/sdks/python/test-suites/direct/common.gradle
index 1dd15ecb09f..10bff26afa5 100644
--- a/sdks/python/test-suites/direct/common.gradle
+++ b/sdks/python/test-suites/direct/common.gradle
@@ -438,17 +438,28 @@ project.tasks.register("inferencePostCommitIT") {
// Create cross-language tasks for running tests against Java expansion
service(s)
def gcpProject = project.findProperty('gcpProject') ?: 'apache-beam-testing'
+def testDirectRunnerOpts = [
+ "--runner=TestDirectRunner",
+ "--project=${gcpProject}",
+ "--temp_location=gs://temp-storage-for-end-to-end-tests/temp-it",
+]
+
+def prismRunnerOpts = [
+ "--runner=PrismRunner",
+ "--environment_type=LOOPBACK",
+ "--project=${gcpProject}",
+ "--temp_location=gs://temp-storage-for-end-to-end-tests/temp-it",
+]
+
+def usePrismRunnerSuites = ["messagingCrossLanguage"]
+
project(":sdks:python:test-suites:xlang").ext.xlangTasks.each { taskMetadata ->
createCrossLanguageUsingJavaExpansionTask(
name: taskMetadata.name,
expansionProjectPaths: taskMetadata.expansionProjectPaths,
collectMarker: taskMetadata.collectMarker,
numParallelTests: 1,
- pythonPipelineOptions: [
- "--runner=TestDirectRunner",
- "--project=${gcpProject}",
- "--temp_location=gs://temp-storage-for-end-to-end-tests/temp-it",
- ],
+ pythonPipelineOptions: usePrismRunnerSuites.contains(taskMetadata.name)
? prismRunnerOpts : testDirectRunnerOpts,
pytestOptions: [
"--capture=no", // print stdout instantly
"--timeout=4500", // timeout of whole command execution
diff --git a/sdks/standard_expansion_services.yaml
b/sdks/standard_expansion_services.yaml
index d0c7f125c44..cb617d3eba1 100644
--- a/sdks/standard_expansion_services.yaml
+++ b/sdks/standard_expansion_services.yaml
@@ -62,6 +62,10 @@
name: 'WriteToMqtt'
'beam:schematransform:org.apache.beam:mqtt_read:v1':
name: 'ReadFromMqtt'
+ 'beam:schematransform:org.apache.beam:jms_write:v1':
+ name: 'WriteToJms'
+ 'beam:schematransform:org.apache.beam:jms_read:v1':
+ name: 'ReadFromJms'
skip_transforms:
# Core SchemaTransforms bundled via :sdks:java:expansion-service; already
# generated from the Java IO expansion service above.
diff --git a/sdks/standard_external_transforms.yaml
b/sdks/standard_external_transforms.yaml
index a3f7d5911db..b589448d64b 100644
--- a/sdks/standard_external_transforms.yaml
+++ b/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-06-11
+# Last updated on: 2026-07-22
- default_service: sdks:java:io:expansion-service:shadowJar
description: ''
@@ -214,6 +214,107 @@
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 a JMS broker and outputs each message as a
single
+ `payload` (string) field.
+
+
+ By default the read is unbounded (streaming): it keeps consuming messages
from
+ the specified queue or topic until the pipeline is stopped. Setting
`maxNumRecords`
+ and/or `maxReadTimeSeconds` bounds the read, producing a bounded (batch)
PCollection.'
+ destinations:
+ python: apache_beam/io
+ fields:
+ - description: 'The JMS acknowledge mode: CLIENT_ACKNOWLEDGE,
CLIENT_ACKNOWLEDGE_UNSAFE,
+ or INDIVIDUAL_ACKNOWLEDGE.'
+ name: acknowledge_mode
+ nullable: true
+ type: str
+ - description: Close timeout for the JMS connection in seconds.
+ name: close_timeout_seconds
+ nullable: true
+ type: int64
+ - description: 'Configuration options to set up the JMS connection.
+
+ Note: if connection factory class name is set, spin up a persistent
expansion
+ service with the provider client JAR on the classpath:
+
+ java -cp <expansion-service-jar>:<provider-client-jar>
+
+ org.apache.beam.sdk.expansion.service.ExpansionService <port>
+
+ and pass expansion_service=''localhost:<port>'' to the transform.
Currently,
+ only ActiveMQ (org.apache.activemq.ActiveMQConnectionFactory) is
embedded into
+ the expansion service'
+ name: connection_configuration
+ nullable: false
+ type: Row(connection_factory_class_name=typing.Optional[str],
password=typing.Optional[str],
+ server_uri=<class 'str'>, username=typing.Optional[str])
+ - description: The proprietary integer code for individual message
acknowledgment
+ when using INDIVIDUAL_ACKNOWLEDGE.
+ name: individual_acknowledge_mode_code
+ nullable: true
+ type: int32
+ - 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
+ - description: The JMS queue to read from. Exclusively one of queue or topic
must
+ be specified.
+ name: queue
+ nullable: true
+ type: str
+ - description: The JMS topic to read from. Exclusively one of queue or topic
must
+ be specified.
+ name: topic
+ nullable: true
+ type: str
+ identifier: beam:schematransform:org.apache.beam:jms_read:v1
+ name: ReadFromJms
+- default_service: sdks:java:io:messaging-expansion-service:shadowJar
+ description: 'Publishes messages to a JMS broker. Expects an input
PCollection of
+ rows with a `payload` (string) or `bytes` field, each of which is
published as
+ one JMS TextMessage.
+
+
+ Works with both bounded (batch) and unbounded (streaming) input
PCollections.'
+ destinations:
+ python: apache_beam/io
+ fields:
+ - description: 'Configuration options to set up the JMS connection.
+
+ Note: if connection factory class name is set, spin up a persistent
expansion
+ service with the provider client JAR on the classpath:
+
+ java -cp <expansion-service-jar>:<provider-client-jar>
+
+ org.apache.beam.sdk.expansion.service.ExpansionService <port>
+
+ and pass expansion_service=''localhost:<port>'' to the transform.
Currently,
+ only ActiveMQ (org.apache.activemq.ActiveMQConnectionFactory) is
embedded into
+ the expansion service'
+ name: connection_configuration
+ nullable: false
+ type: Row(connection_factory_class_name=typing.Optional[str],
password=typing.Optional[str],
+ server_uri=<class 'str'>, username=typing.Optional[str])
+ - description: The JMS queue to write to. Exclusively one of queue or topic
must
+ be specified.
+ name: queue
+ nullable: true
+ type: str
+ - description: The JMS topic to write to. Exclusively one of queue or topic
must
+ be specified.
+ name: topic
+ nullable: true
+ type: str
+ identifier: beam:schematransform:org.apache.beam:jms_write:v1
+ name: WriteToJms
- 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.
