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 e0266d62f2e0..f1ba03a243ee 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 3949d1af3248..2b502c679db9 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 8926d584a133..940248773202 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.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.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 static Write write() { return new AutoValue_JmsIO_Write.Builder().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 withConnectionFactory(ConnectionFactory connectionFactory); @@ -367,6 +482,21 @@ public Read withConnectionFactoryProviderFn( return builder().setConnectionFactoryProviderFn(connectionFactoryProviderFn).build(); } + public Read withConnectionConfiguration(ConnectionConfiguration configuration) { + checkArgument(configuration != null, "configuration can not be null"); + Read read = + this.withConnectionFactoryProviderFn( + (SerializableFunction) + __ -> 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 Write withConnectionFactoryProviderFn( return builder().setConnectionFactoryProviderFn(connectionFactoryProviderFn).build(); } + public Write withConnectionConfiguration(ConnectionConfiguration configuration) { + checkArgument(configuration != null, "configuration can not be null"); + Write write = + this.withConnectionFactoryProviderFn( + (SerializableFunction) + __ -> 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 000000000000..79643da33279 --- /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 { + @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 :\n" + + "org.apache.beam.sdk.expansion.service.ExpansionService \n" + + "and pass expansion_service='localhost:' 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 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 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 outputRows = + input + .getPipeline() + .apply(readTransform) + .apply( + "Wrap in Beam Rows", + ParDo.of( + new DoFn() { + @ProcessElement + public void processElement( + @Element JmsRecord record, OutputReceiver 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 000000000000..5db7271b4b98 --- /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 { + @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 :\n" + + "org.apache.beam.sdk.expansion.service.ExpansionService \n" + + "and pass expansion_service='localhost:' 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 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 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 writeTransform = + JmsIO.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() { + @ProcessElement + public void processElement( + @Element Row row, OutputReceiver 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 000000000000..b354ef94a008 --- /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 serviceLoader = + ServiceLoader.load(SchemaTransformProvider.class); + List 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 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 serviceLoader = + ServiceLoader.load(SchemaTransformProvider.class); + List 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 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 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 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 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 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 6cdf67b86d6e..8693f4597bc1 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 000000000000..98905107f05e --- /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)]) + + +@pytest.mark.uses_messaging_java_expansion_service +@unittest.skipIf( + DockerContainer is None, 'testcontainers package is not installed') +@unittest.skipIf( + ReadFromJms is None or WriteToJms is None, + 'JMS cross-language wrappers are not generated') +@unittest.skipIf( + TestPipeline().get_pipeline_options().view_as(StandardOptions).runner + is None, + 'Do not run this test on precommit suites.') +@unittest.skipIf( + '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 889096d1f22b..dad85a688ecb 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 @@ def subscribe(): 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 1dd15ecb09f9..10bff26afa5d 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 d0c7f125c44b..cb617d3eba12 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 a3f7d5911dbb..b589448d64b5 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 : + + org.apache.beam.sdk.expansion.service.ExpansionService + + and pass expansion_service=''localhost:'' 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=, 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 : + + org.apache.beam.sdk.expansion.service.ExpansionService + + and pass expansion_service=''localhost:'' 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=, 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.