diff --git a/.github/workflows/beam_PreCommit_Java_Jms_IO_Direct.yml b/.github/workflows/beam_PreCommit_Java_Jms_IO_Direct.yml index d032b22d3527..597d52a6d1b2 100644 --- a/.github/workflows/beam_PreCommit_Java_Jms_IO_Direct.yml +++ b/.github/workflows/beam_PreCommit_Java_Jms_IO_Direct.yml @@ -85,6 +85,8 @@ jobs: github_job: ${{ matrix.job_name }} (${{ matrix.job_phrase }}) - name: Setup environment uses: ./.github/actions/setup-environment-action + with: + java-version: 17 - name: run Jms IO build script uses: ./.github/actions/gradle-command-self-hosted-action with: diff --git a/CHANGES.md b/CHANGES.md index bd9c6ea246a5..a1f6a3154eae 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -82,6 +82,9 @@ * (Python) Typehints of dataclass fields are honored during type inferences. To restore the behavior of fallback-to-any, use pipeline option `--exclude_infer_dataclass_field_type` ([#38797](https://github.com/apache/beam/issues/38797)). However fixing forward is recommended. +* (Java) JmsIO migrated to `jakarta.jms` (JMS 3.1) and ActiveMQ 6.2.5. User code implementing `JmsIO.MessageMapper`, + `valueMapper`, `topicNameMapper`, or providing a `ConnectionFactory` must update imports from `javax.jms.*` to + `jakarta.jms.*`. The module now requires Java 17 at runtime ([#38729](https://github.com/apache/beam/issues/38729)). ## Deprecations diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy index 344cf66e01fc..b1ab2ca1e333 100644 --- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy +++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy @@ -598,6 +598,7 @@ class BeamModulePlugin implements Plugin { // There are a few versions are determined by the BOMs by running scripts/tools/bomupgrader.py // marked as [bomupgrader]. See the documentation of that script for detail. def activemq_version = "5.19.2" + def activemq6_version = "6.2.5" def autovalue_version = "1.9" def autoservice_version = "1.0.1" def aws_java_sdk2_version = "2.20.162" @@ -639,7 +640,7 @@ class BeamModulePlugin implements Plugin { def postgres_version = "42.6.2" // [bomupgrader] determined by: com.google.protobuf:protobuf-java, consistent with: google_cloud_platform_libraries_bom def protobuf_version = "4.33.2" - def qpid_jms_client_version = "0.61.0" + def qpid_jms_client_version = "2.10.0" def quickcheck_version = "1.0" def sbe_tool_version = "1.25.1" def singlestore_jdbc_version = "1.1.4" @@ -678,6 +679,11 @@ class BeamModulePlugin implements Plugin { activemq_junit : "org.apache.activemq.tooling:activemq-junit:$activemq_version", activemq_kahadb_store : "org.apache.activemq:activemq-kahadb-store:$activemq_version", activemq_mqtt : "org.apache.activemq:activemq-mqtt:$activemq_version", + activemq6_amqp : "org.apache.activemq:activemq-amqp:$activemq6_version", + activemq6_broker : "org.apache.activemq:activemq-broker:$activemq6_version", + activemq6_client : "org.apache.activemq:activemq-client:$activemq6_version", + activemq6_jaas : "org.apache.activemq:activemq-jaas:$activemq6_version", + activemq6_kahadb_store : "org.apache.activemq:activemq-kahadb-store:$activemq6_version", aircompressor : "io.airlift:aircompressor:2.0.3", args4j : "args4j:args4j:2.33", auto_value_annotations : "com.google.auto.value:auto-value-annotations:$autovalue_version", diff --git a/sdks/java/io/amqp/build.gradle b/sdks/java/io/amqp/build.gradle index 6f2899eeb05b..9333a8e9d21f 100644 --- a/sdks/java/io/amqp/build.gradle +++ b/sdks/java/io/amqp/build.gradle @@ -19,6 +19,19 @@ plugins { id 'org.apache.beam.module' } applyJavaNature( automaticModuleName: 'org.apache.beam.sdk.io.amqp') +// The JMS IO module pulls ActiveMQ 6.2.5 (Java 17) into the shared dependency +// map. Because the global resolution strategy forces every mapped version, +// that 6.2.5 would otherwise be forced onto this Java 11 module via the shared +// org.apache.activemq coordinates. AMQP IO only needs ActiveMQ as an embedded +// test broker, so pin it back to the Java 11-compatible 5.x line. +configurations.all { + resolutionStrategy.dependencySubstitution { + substitute module('org.apache.activemq:activemq-broker') using module(library.java.activemq_broker) + substitute module('org.apache.activemq:activemq-amqp') using module(library.java.activemq_amqp) + substitute module('org.apache.activemq:activemq-client') using module(library.java.activemq_client) + } +} + description = "Apache Beam :: SDKs :: Java :: IO :: AMQP" ext.summary = "IO to read and write using AMQP 1.0 protocol (http://www.amqp.org)." diff --git a/sdks/java/io/jms/build.gradle b/sdks/java/io/jms/build.gradle index 24a195e63f14..4b038c6bb28d 100644 --- a/sdks/java/io/jms/build.gradle +++ b/sdks/java/io/jms/build.gradle @@ -19,6 +19,7 @@ plugins { id 'org.apache.beam.module' } applyJavaNature( automaticModuleName: 'org.apache.beam.sdk.io.jms', + requireJavaVersion: JavaVersion.VERSION_17, ) provideIntegrationTestingDependencies() enableJavaPerformanceTesting() @@ -32,12 +33,12 @@ dependencies { implementation project(path: ":sdks:java:core", configuration: "shadow") implementation library.java.slf4j_api implementation library.java.joda_time - implementation "org.apache.geronimo.specs:geronimo-jms_2.0_spec:1.0-alpha-2" - testImplementation library.java.activemq_amqp - testImplementation library.java.activemq_broker - testImplementation library.java.activemq_jaas - testImplementation library.java.activemq_kahadb_store - testImplementation library.java.activemq_client + implementation "jakarta.jms:jakarta.jms-api:3.1.0" + testImplementation library.java.activemq6_amqp + testImplementation library.java.activemq6_broker + testImplementation library.java.activemq6_jaas + testImplementation library.java.activemq6_kahadb_store + testImplementation library.java.activemq6_client testImplementation library.java.hamcrest testImplementation library.java.junit testImplementation library.java.mockito_core diff --git a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java index 276d61ae5ef5..5780235c5b54 100644 --- a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java +++ b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java @@ -17,14 +17,14 @@ */ package org.apache.beam.sdk.io.jms; +import jakarta.jms.JMSException; +import jakarta.jms.Message; +import jakarta.jms.MessageConsumer; +import jakarta.jms.Session; import java.io.IOException; import java.io.Serializable; import java.util.Objects; import java.util.concurrent.locks.ReentrantReadWriteLock; -import javax.jms.JMSException; -import javax.jms.Message; -import javax.jms.MessageConsumer; -import javax.jms.Session; import org.apache.beam.sdk.io.UnboundedSource; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; import org.checkerframework.checker.nullness.qual.Nullable; 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 5deb7a2b62c7..d1c666a83ad7 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 @@ -22,6 +22,15 @@ import com.google.auto.value.AutoValue; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import jakarta.jms.Connection; +import jakarta.jms.ConnectionFactory; +import jakarta.jms.Destination; +import jakarta.jms.JMSException; +import jakarta.jms.Message; +import jakarta.jms.MessageConsumer; +import jakarta.jms.MessageProducer; +import jakarta.jms.Session; +import jakarta.jms.TextMessage; import java.io.IOException; import java.io.Serializable; import java.nio.charset.StandardCharsets; @@ -36,15 +45,6 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.stream.Stream; -import javax.jms.Connection; -import javax.jms.ConnectionFactory; -import javax.jms.Destination; -import javax.jms.JMSException; -import javax.jms.Message; -import javax.jms.MessageConsumer; -import javax.jms.MessageProducer; -import javax.jms.Session; -import javax.jms.TextMessage; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.SerializableCoder; import org.apache.beam.sdk.io.Read.Unbounded; @@ -86,9 +86,9 @@ * *

JmsIO source returns unbounded collection of JMS records as {@code PCollection}. A * {@link JmsRecord} includes JMS headers and properties, along with the JMS {@link - * javax.jms.TextMessage} payload. + * jakarta.jms.TextMessage} payload. * - *

To configure a JMS source, you have to provide a {@link javax.jms.ConnectionFactory} and the + *

To configure a JMS source, you have to provide a {@link jakarta.jms.ConnectionFactory} and the * destination (queue or topic) where to consume. The following example illustrates various options * for configuring the source: * @@ -102,8 +102,8 @@ * * } * - *

It is possible to read any type of JMS {@link javax.jms.Message} into a custom POJO using the - * following configuration: + *

It is possible to read any type of JMS {@link jakarta.jms.Message} into a custom POJO using + * the following configuration: * *

{@code
  * pipeline.apply(JmsIO.readMessage()
@@ -121,8 +121,8 @@
  * 

Writing to a JMS destination

* *

JmsIO sink supports writing text messages to a JMS destination on a broker. To configure a JMS - * sink, you must specify a {@link javax.jms.ConnectionFactory} and a {@link javax.jms.Destination} - * name. For instance: + * sink, you must specify a {@link jakarta.jms.ConnectionFactory} and a {@link + * jakarta.jms.Destination} name. For instance: * *

{@code
  * pipeline
@@ -1053,7 +1053,7 @@ public Write withTopicNameMapper(SerializableFunction to
     }
 
     /**
-     * Map the {@code EventT} object to a {@link javax.jms.Message}.
+     * Map the {@code EventT} object to a {@link jakarta.jms.Message}.
      *
      * 

For instance: * @@ -1075,7 +1075,7 @@ public Write withTopicNameMapper(SerializableFunction to * .apply(JmsIO.write().withValueMapper(valueNapper) * }

* - * @param valueMapper The function returning the {@link javax.jms.Message} + * @param valueMapper The function returning the {@link jakarta.jms.Message} * @return The corresponding {@link JmsIO.Write}. */ public Write withValueMapper( diff --git a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsRecord.java b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsRecord.java index 9f547a801d04..137ec39fe715 100644 --- a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsRecord.java +++ b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsRecord.java @@ -18,10 +18,10 @@ package org.apache.beam.sdk.io.jms; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import jakarta.jms.Destination; import java.io.Serializable; import java.util.Map; import java.util.Objects; -import javax.jms.Destination; import org.checkerframework.checker.nullness.qual.Nullable; /** diff --git a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/TextMessageMapper.java b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/TextMessageMapper.java index d5d85c46794d..6a93c5ea639f 100644 --- a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/TextMessageMapper.java +++ b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/TextMessageMapper.java @@ -17,15 +17,15 @@ */ package org.apache.beam.sdk.io.jms; -import javax.jms.JMSException; -import javax.jms.Message; -import javax.jms.Session; -import javax.jms.TextMessage; +import jakarta.jms.JMSException; +import jakarta.jms.Message; +import jakarta.jms.Session; +import jakarta.jms.TextMessage; import org.apache.beam.sdk.transforms.SerializableBiFunction; /** - * The TextMessageMapper takes a {@link String} value, a {@link javax.jms.Session} and returns a - * {@link javax.jms.TextMessage}. + * The TextMessageMapper takes a {@link String} value, a {@link jakarta.jms.Session} and returns a + * {@link jakarta.jms.TextMessage}. */ public class TextMessageMapper implements SerializableBiFunction { diff --git a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/CommonJms.java b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/CommonJms.java index 1d1245e6877d..b8d6cf23fae6 100644 --- a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/CommonJms.java +++ b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/CommonJms.java @@ -17,15 +17,15 @@ */ package org.apache.beam.sdk.io.jms; +import jakarta.jms.BytesMessage; +import jakarta.jms.ConnectionFactory; +import jakarta.jms.Message; import java.io.Serializable; import java.lang.reflect.InvocationTargetException; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.List; import java.util.function.Supplier; -import javax.jms.BytesMessage; -import javax.jms.ConnectionFactory; -import javax.jms.Message; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.broker.BrokerPlugin; import org.apache.activemq.broker.BrokerService; @@ -143,7 +143,7 @@ Class getConnectionFactoryClass() { return this.connectionFactoryClass; } - /** A test class that maps a {@link javax.jms.BytesMessage} into a {@link String}. */ + /** A test class that maps a {@link jakarta.jms.BytesMessage} into a {@link String}. */ public static class BytesMessageToStringMessageMapper implements JmsIO.MessageMapper { @Override diff --git a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOIT.java b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOIT.java index 266d04342d1f..e4f484c148d7 100644 --- a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOIT.java +++ b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOIT.java @@ -23,6 +23,13 @@ import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertTrue; +import jakarta.jms.Connection; +import jakarta.jms.ConnectionFactory; +import jakarta.jms.JMSException; +import jakarta.jms.Message; +import jakarta.jms.QueueBrowser; +import jakarta.jms.Session; +import jakarta.jms.TextMessage; import java.io.IOException; import java.io.Serializable; import java.time.Instant; @@ -32,13 +39,6 @@ import java.util.Set; import java.util.UUID; import java.util.function.Function; -import javax.jms.Connection; -import javax.jms.ConnectionFactory; -import javax.jms.JMSException; -import javax.jms.Message; -import javax.jms.QueueBrowser; -import javax.jms.Session; -import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.command.ActiveMQTextMessage; import org.apache.beam.sdk.PipelineResult; diff --git a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java index b3233f866172..6bc425ca3e21 100644 --- a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java +++ b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java @@ -51,6 +51,16 @@ import static org.mockito.Mockito.verifyNoMoreInteractions; import static org.mockito.Mockito.when; +import jakarta.jms.BytesMessage; +import jakarta.jms.Connection; +import jakarta.jms.ConnectionFactory; +import jakarta.jms.JMSException; +import jakarta.jms.Message; +import jakarta.jms.MessageConsumer; +import jakarta.jms.MessageProducer; +import jakarta.jms.QueueBrowser; +import jakarta.jms.Session; +import jakarta.jms.TextMessage; import java.io.IOException; import java.io.NotSerializableException; import java.io.Serializable; @@ -71,16 +81,6 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; -import javax.jms.BytesMessage; -import javax.jms.Connection; -import javax.jms.ConnectionFactory; -import javax.jms.JMSException; -import javax.jms.Message; -import javax.jms.MessageConsumer; -import javax.jms.MessageProducer; -import javax.jms.QueueBrowser; -import javax.jms.Session; -import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.command.ActiveMQMessage; import org.apache.activemq.util.Callback; diff --git a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/MockNonSerializableConnectionFactory.java b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/MockNonSerializableConnectionFactory.java index 752123327e9b..b9deef4991e7 100644 --- a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/MockNonSerializableConnectionFactory.java +++ b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/MockNonSerializableConnectionFactory.java @@ -17,10 +17,10 @@ */ package org.apache.beam.sdk.io.jms; -import javax.jms.Connection; -import javax.jms.ConnectionFactory; -import javax.jms.JMSContext; -import javax.jms.JMSException; +import jakarta.jms.Connection; +import jakarta.jms.ConnectionFactory; +import jakarta.jms.JMSContext; +import jakarta.jms.JMSException; public class MockNonSerializableConnectionFactory implements ConnectionFactory { @Override diff --git a/sdks/java/io/mqtt/build.gradle b/sdks/java/io/mqtt/build.gradle index 85d6c6c16f25..a088c0447410 100644 --- a/sdks/java/io/mqtt/build.gradle +++ b/sdks/java/io/mqtt/build.gradle @@ -19,6 +19,19 @@ plugins { id 'org.apache.beam.module' } applyJavaNature( automaticModuleName: 'org.apache.beam.sdk.io.mqtt') +// The JMS IO module pulls ActiveMQ 6.2.5 (Java 17) into the shared dependency +// map. Because the global resolution strategy forces every mapped version, +// that 6.2.5 would otherwise be forced onto this Java 11 module via the shared +// org.apache.activemq coordinates. MQTT IO only needs ActiveMQ as an embedded +// test broker, so pin it back to the Java 11-compatible 5.x line. +configurations.all { + resolutionStrategy.dependencySubstitution { + substitute module('org.apache.activemq:activemq-broker') using module(library.java.activemq_broker) + substitute module('org.apache.activemq:activemq-client') using module(library.java.activemq_client) + substitute module('org.apache.activemq:activemq-kahadb-store') using module(library.java.activemq_kahadb_store) + } +} + description = "Apache Beam :: SDKs :: Java :: IO :: MQTT" ext.summary = "IO to read and write to a MQTT broker."