diff --git a/CHANGES.md b/CHANGES.md index eb374150cadf..cb5eced42f7c 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -76,8 +76,10 @@ ## Breaking Changes -* X behavior was changed ([#X](https://github.com/apache/beam/issues/X)). * (Go) The row coder now encodes `int16` and `uint16` struct fields as 2 byte big endian INT16 values, matching the Java and Python SDKs. This is an update incompatible change for streaming pipelines that use rows with `int16` or `uint16` fields ([#40151](https://github.com/apache/beam/issues/40151)). +* (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 1453b63b2f2f..8994376e9afe 100644 --- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy +++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy @@ -608,7 +608,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.5" + def activemq_version = "6.2.5" def autovalue_version = "1.9" def autoservice_version = "1.0.1" def aws_java_sdk2_version = "2.20.162" @@ -650,7 +650,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.6" - 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" diff --git a/sdks/java/io/amqp/build.gradle b/sdks/java/io/amqp/build.gradle index 6f2899eeb05b..440ff895f737 100644 --- a/sdks/java/io/amqp/build.gradle +++ b/sdks/java/io/amqp/build.gradle @@ -19,6 +19,22 @@ 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. +def activemq5_version = '5.19.5' +configurations.all { + resolutionStrategy.dependencySubstitution { + substitute module('org.apache.activemq:activemq-broker') using module("org.apache.activemq:activemq-broker:${activemq5_version}") + substitute module('org.apache.activemq:activemq-amqp') using module("org.apache.activemq:activemq-amqp:${activemq5_version}") + substitute module('org.apache.activemq:activemq-client') using module("org.apache.activemq:activemq-client:${activemq5_version}") + substitute module('org.apache.activemq:activemq-kahadb-store') using module("org.apache.activemq:activemq-kahadb-store:${activemq5_version}") + substitute module('org.apache.activemq.tooling:activemq-junit') using module("org.apache.activemq.tooling:activemq-junit:${activemq5_version}") + } +} + 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 369d02b37d3e..e801701caa07 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,13 +33,10 @@ 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" // Don't put proprietary licensed ibm mq client into runtimeClasspath // (affects expansion service shadow jar) - compileOnly("com.ibm.mq:com.ibm.mq.allclient:9.3.0.25") { - // duplicating geronimo-jms_2.0_spec - exclude group: "javax.jms", module: "javax.jms-api" - } + compileOnly "com.ibm.mq:com.ibm.mq.jakarta.client:9.3.0.25" + implementation "jakarta.jms:jakarta.jms-api:3.1.0" testImplementation library.java.activemq_amqp testImplementation library.java.activemq_broker testImplementation library.java.activemq_jaas diff --git a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/BeamGenericJmsConnectionFactory.java b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/BeamGenericJmsConnectionFactory.java index d14c8c8addc3..695fefdc1ac1 100644 --- a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/BeamGenericJmsConnectionFactory.java +++ b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/BeamGenericJmsConnectionFactory.java @@ -17,8 +17,8 @@ */ package org.apache.beam.sdk.io.jms; +import jakarta.jms.ConnectionFactory; import java.io.Serializable; -import javax.jms.ConnectionFactory; /** * An interface for creating custom JMS {@link ConnectionFactory} instances. diff --git a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/ConnectionConfiguration.java b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/ConnectionConfiguration.java index f534f8a07313..cfa94a739b25 100644 --- a/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/ConnectionConfiguration.java +++ b/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/ConnectionConfiguration.java @@ -20,14 +20,14 @@ import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; import com.google.auto.value.AutoValue; -import com.ibm.mq.jms.MQConnectionFactory; -import com.ibm.msg.client.wmq.WMQConstants; +import com.ibm.mq.jakarta.jms.MQConnectionFactory; +import com.ibm.msg.client.jakarta.wmq.WMQConstants; +import jakarta.jms.ConnectionFactory; import java.io.Serializable; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.net.URI; import java.util.List; -import javax.jms.ConnectionFactory; import org.apache.beam.sdk.schemas.AutoValueSchema; import org.apache.beam.sdk.schemas.annotations.DefaultSchema; import org.apache.beam.sdk.schemas.annotations.SchemaFieldDescription; 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 1a00216564ae..fdfd2c0c7bce 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,6 +17,10 @@ */ 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.ArrayList; @@ -24,10 +28,6 @@ import java.util.Objects; import java.util.concurrent.atomic.AtomicInteger; 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.sdk.io.jms.JmsIO.AcknowledgeMode; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; 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 7fafe074a9f1..8de0ef52698a 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; @@ -37,15 +46,6 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; 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; @@ -87,9 +87,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: * @@ -103,8 +103,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()
@@ -122,10 +122,10 @@
  * 

Acknowledgment Modes and Client Prefetch Configuration

* *

By default, {@link JmsIO} consumes messages using {@link AcknowledgeMode#CLIENT_ACKNOWLEDGE} - * where a new {@link javax.jms.Session} is created for each checkpoint to prevent premature + * where a new {@link jakarta.jms.Session} is created for each checkpoint to prevent premature * acknowledgments across bundles. When using {@link AcknowledgeMode#CLIENT_ACKNOWLEDGE}, if your * JMS broker or client library utilizes client-side message prefetch buffers (such as Apache - * ActiveMQ), you should configure {@code prefetch=0} on your {@link javax.jms.ConnectionFactory} + * ActiveMQ), you should configure {@code prefetch=0} on your {@link jakarta.jms.ConnectionFactory} * (e.g., via {@code ?jms.prefetchPolicy.all=0} in the broker URL or {@code * ActiveMQPrefetchPolicy.setAll(0)}). Otherwise, unconsumed messages could be held inside old * consumers in low throughput scenario and could lead to message backlog. @@ -138,8 +138,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
@@ -1223,7 +1223,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: * @@ -1245,7 +1245,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 749b215b2059..106d21ead619 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.broker.BrokerPlugin; import org.apache.activemq.broker.BrokerService; import org.apache.activemq.security.AuthenticationUser; @@ -160,7 +160,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/ConnectionConfigurationTest.java b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/ConnectionConfigurationTest.java index 7aabc683855d..11db2e9f7168 100644 --- a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/ConnectionConfigurationTest.java +++ b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/ConnectionConfigurationTest.java @@ -22,9 +22,9 @@ import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; -import javax.jms.Connection; -import javax.jms.ConnectionFactory; -import javax.jms.JMSException; +import jakarta.jms.Connection; +import jakarta.jms.ConnectionFactory; +import jakarta.jms.JMSException; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.qpid.jms.JmsConnectionFactory; import org.checkerframework.checker.nullness.qual.Nullable; @@ -79,22 +79,22 @@ public Connection createConnection(String username, String password) throws JMSE } @Override - public javax.jms.JMSContext createContext() { + public jakarta.jms.JMSContext createContext() { return null; } @Override - public javax.jms.JMSContext createContext(int sessionMode) { + public jakarta.jms.JMSContext createContext(int sessionMode) { return null; } @Override - public javax.jms.JMSContext createContext(String username, String password) { + public jakarta.jms.JMSContext createContext(String username, String password) { return null; } @Override - public javax.jms.JMSContext createContext(String username, String password, int sessionMode) { + public jakarta.jms.JMSContext createContext(String username, String password, int sessionMode) { return null; } } @@ -137,11 +137,12 @@ public void testCustomBeamGenericJmsConnectionFactory() { public void testClientsNotSlippedIntoRuntimeDependencies() { // Verify IBM MQ client is not present on runtime classpath assertThrows( - ClassNotFoundException.class, () -> Class.forName("com.ibm.mq.jms.MQConnectionFactory")); + ClassNotFoundException.class, + () -> Class.forName("com.ibm.mq.jakarta.jms.MQConnectionFactory")); ConnectionConfiguration config = ConnectionConfiguration.create("tcp://localhost:1414") - .withConnectionFactoryClassName("com.ibm.mq.jms.MQConnectionFactory"); + .withConnectionFactoryClassName("com.ibm.mq.jakarta.jms.MQConnectionFactory"); IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, config::createConnectionFactory); assertTrue(exception.getCause() instanceof ClassNotFoundException); 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 3dbb20775f71..edbb41a7ac79 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.assertEquals; import static org.junit.Assert.assertNotEquals; +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; @@ -34,13 +41,6 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; 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 d6cb134249ae..c92794c21e58 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 @@ -45,6 +45,16 @@ import static org.mockito.Mockito.verify; 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.Serializable; import java.lang.reflect.Proxy; @@ -61,16 +71,6 @@ import java.util.Set; 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; @@ -568,8 +568,8 @@ public void testCheckpointMarkAndFinalizeSeparately() throws Exception { assertEquals(1, jmsMark.getMessages().size()); // consume two more messages after checkpoint made - reader.advance(); - reader.advance(); + assertTrue(advanceWithRetry(reader)); + assertTrue(advanceWithRetry(reader)); // the messages are still pending in the queue (no ACK yet) assertEquals(10, count(QUEUE)); @@ -596,8 +596,8 @@ public void testCheckpointMarkAndFinalizeSeparatelyIndividualAcknowledge() throw assertNotNull(jmsMark.getMessages()); assertEquals(3, jmsMark.getMessages().size()); - reader.advance(); - reader.advance(); + assertTrue(advanceWithRetry(reader)); + assertTrue(advanceWithRetry(reader)); assertEquals(10, count(QUEUE)); mark.finalizeCheckpoint(); @@ -621,8 +621,8 @@ public void testCheckpointMarkAndFinalizeSeparatelyClientAcknowledgeUnsafe() thr assertNotNull(jmsMark.getMessages()); assertEquals(1, jmsMark.getMessages().size()); - reader.advance(); - reader.advance(); + assertTrue(advanceWithRetry(reader)); + assertTrue(advanceWithRetry(reader)); assertEquals(10, count(QUEUE)); mark.finalizeCheckpoint(); diff --git a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java index 4ea8f6d317a7..277c84151703 100644 --- a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java +++ b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsLocalTest.java @@ -26,19 +26,19 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import jakarta.jms.Connection; +import jakarta.jms.ConnectionFactory; +import jakarta.jms.JMSException; +import jakarta.jms.Message; +import jakarta.jms.MessageConsumer; +import jakarta.jms.Queue; +import jakarta.jms.Session; import java.io.IOException; import java.io.NotSerializableException; import java.util.List; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; -import javax.jms.Connection; -import javax.jms.ConnectionFactory; -import javax.jms.JMSException; -import javax.jms.Message; -import javax.jms.MessageConsumer; -import javax.jms.Queue; -import javax.jms.Session; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.options.ExecutorOptions; 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/messaging-expansion-service/build.gradle b/sdks/java/io/messaging-expansion-service/build.gradle index 8693f4597bc1..5d30a9270c06 100644 --- a/sdks/java/io/messaging-expansion-service/build.gradle +++ b/sdks/java/io/messaging-expansion-service/build.gradle @@ -22,6 +22,7 @@ mainClassName = "org.apache.beam.sdk.expansion.service.ExpansionService" applyJavaNature( automaticModuleName: 'org.apache.beam.sdk.io.messaging.expansion.service', + requireJavaVersion: JavaVersion.VERSION_17, exportJavadoc: false, validateShadowJar: false, shadowClosure: {}, diff --git a/sdks/java/io/mqtt/build.gradle b/sdks/java/io/mqtt/build.gradle index 1f2d808ed7e0..b828632165e5 100644 --- a/sdks/java/io/mqtt/build.gradle +++ b/sdks/java/io/mqtt/build.gradle @@ -19,6 +19,21 @@ 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. +def activemq5_version = '5.19.5' +configurations.all { + resolutionStrategy.dependencySubstitution { + substitute module('org.apache.activemq:activemq-broker') using module("org.apache.activemq:activemq-broker:${activemq5_version}") + substitute module('org.apache.activemq:activemq-client') using module("org.apache.activemq:activemq-client:${activemq5_version}") + substitute module('org.apache.activemq:activemq-kahadb-store') using module("org.apache.activemq:activemq-kahadb-store:${activemq5_version}") + substitute module('org.apache.activemq:activemq-mqtt') using module("org.apache.activemq:activemq-mqtt:${activemq5_version}") + } +} + description = "Apache Beam :: SDKs :: Java :: IO :: MQTT" ext.summary = "IO to read and write to a MQTT broker." 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 index c3f260973765..3f926f1766c8 100644 --- a/sdks/python/apache_beam/io/external/xlang_jmsio_it_test.py +++ b/sdks/python/apache_beam/io/external/xlang_jmsio_it_test.py @@ -306,7 +306,7 @@ def start_ibm_mq_container(cls, retries): cls.expansion_service_obj = BeamJarExpansionService( 'sdks:java:io:messaging-expansion-service:shadowJar', classpath=[ - 'com.ibm.mq:com.ibm.mq.allclient:9.3.0.25', + 'com.ibm.mq:com.ibm.mq.jakarta.client:9.3.0.25', 'org.json:json:20251224' ]) cls.expansion_service = cls.expansion_service_obj.__enter__() @@ -327,7 +327,7 @@ def _connection_configuration(self, connection_param=None): uri += '&' + connection_param return { 'server_uri': uri, - 'connection_factory_class_name': 'com.ibm.mq.jms.MQConnectionFactory', + 'connection_factory_class_name': 'com.ibm.mq.jakarta.jms.MQConnectionFactory', 'username': 'app', 'password': 'admin123' } diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index 54be607eabab..becfaa793032 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -1163,7 +1163,7 @@ def temp_ibm_mq_server(): yield { 'SERVER_URI': f'tcp://{host}:{port}?channel=DEV.APP.SVRCONN&queueManager=QM1', - 'CONNECTION_FACTORY_CLASS_NAME': 'com.ibm.mq.jms.MQConnectionFactory', + 'CONNECTION_FACTORY_CLASS_NAME': 'com.ibm.mq.jakarta.jms.MQConnectionFactory', 'USERNAME': 'app', 'PASSWORD': 'admin123', 'SOURCE_QUEUE': 'DEV.QUEUE.1', diff --git a/sdks/python/apache_beam/yaml/standard_io.yaml b/sdks/python/apache_beam/yaml/standard_io.yaml index 15aa9186c0fc..fb26bfb6d350 100644 --- a/sdks/python/apache_beam/yaml/standard_io.yaml +++ b/sdks/python/apache_beam/yaml/standard_io.yaml @@ -276,7 +276,7 @@ config: gradle_target: 'sdks:java:io:messaging-expansion-service:shadowJar' classpath: - - 'com.ibm.mq:com.ibm.mq.allclient:9.3.0.25' + - 'com.ibm.mq:com.ibm.mq.jakarta.client:9.3.0.25' - 'org.json:json:20251224'