Skip to content

Update JmsIO to ActiveMQ 6.2.5 and jakarta.jms #38729

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jbonofre wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Update JmsIO to ActiveMQ 6.2.5 and jakarta.jms #38729

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .github/workflows/beam_PreCommit_Java_Jms_IO_Direct.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
3 changes: 3 additions & 0 deletions CHANGES.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
8 changes: 7 additions & 1 deletion buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -598,6 +598,7 @@ class BeamModulePlugin implements Plugin<Project> {
// 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"
Expand Down Expand Up @@ -639,7 +640,7 @@ class BeamModulePlugin implements Plugin<Project> {
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"

@gemini-code-assist gemini-code-assist Bot May 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

Qpid JMS Client 2.x requires Java 11 as a minimum runtime version. Since Apache Beam still supports Java 8, upgrading qpid_jms_client_version to 2.10.0 will break Java 8 compatibility for any users or tests utilizing this client. Please verify if this is intentional or if we need to restrict this dependency/module to Java 11+.

def quickcheck_version = "1.0"
def sbe_tool_version = "1.25.1"
def singlestore_jdbc_version = "1.1.4"
Expand Down Expand Up @@ -678,6 +679,11 @@ class BeamModulePlugin implements Plugin<Project> {
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",
Expand Down
13 changes: 13 additions & 0 deletions sdks/java/io/amqp/build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)."

Expand Down
13 changes: 7 additions & 6 deletions sdks/java/io/jms/build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
plugins { id 'org.apache.beam.module' }
applyJavaNature(
automaticModuleName: 'org.apache.beam.sdk.io.jms',
requireJavaVersion: JavaVersion.VERSION_17,
)
provideIntegrationTestingDependencies()
enableJavaPerformanceTesting()
Expand All @@ -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"

@gemini-code-assist gemini-code-assist Bot May 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

jakarta.jms:jakarta.jms-api:3.1.0 (Jakarta Messaging 3.1) is compiled for Java 11 (class file version 55.0). Since Apache Beam still supports Java 8, using version 3.1.0 will break Java 8 compatibility for the beam-sdks-java-io-jms module.

To maintain Java 8 compatibility while migrating to the jakarta.jms namespace, please use jakarta.jms:jakarta.jms-api:3.0.0 (Jakarta Messaging 3.0), which is compiled for Java 8 (class file version 52.0).

  implementation "jakarta.jms:jakarta.jms-api:3.0.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
Expand Down
8 changes: 4 additions & 4 deletions sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
34 changes: 17 additions & 17 deletions sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -86,9 +86,9 @@
*
* <p>JmsIO source returns unbounded collection of JMS records as {@code PCollection<JmsRecord>}. A
* {@link JmsRecord} includes JMS headers and properties, along with the JMS {@link
* javax.jms.TextMessage} payload.
* jakarta.jms.TextMessage} payload.
*
* <p>To configure a JMS source, you have to provide a {@link javax.jms.ConnectionFactory} and the
* <p>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:
*
Expand All @@ -102,8 +102,8 @@
*
* }</pre>
*
* <p>It is possible to read any type of JMS {@link javax.jms.Message} into a custom POJO using the
* following configuration:
* <p>It is possible to read any type of JMS {@link jakarta.jms.Message} into a custom POJO using
* the following configuration:
*
* <pre>{@code
* pipeline.apply(JmsIO.<T>readMessage()
Expand All @@ -121,8 +121,8 @@
* <h3>Writing to a JMS destination</h3>
*
* <p>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:
*
* <pre>{@code
* pipeline
Expand Down Expand Up @@ -1053,7 +1053,7 @@ public Write<EventT> withTopicNameMapper(SerializableFunction<EventT, String> to
}

/**
* Map the {@code EventT} object to a {@link javax.jms.Message}.
* Map the {@code EventT} object to a {@link jakarta.jms.Message}.
*
* <p>For instance:
*
Expand All @@ -1075,7 +1075,7 @@ public Write<EventT> withTopicNameMapper(SerializableFunction<EventT, String> to
* .apply(JmsIO.write().withValueMapper(valueNapper)
* }</pre>
*
* @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<EventT> withValueMapper(
Expand Down
2 changes: 1 addition & 1 deletion sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsRecord.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

/**
Expand Down
12 changes: 6 additions & 6 deletions sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/TextMessageMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<String, Session, Message> {

Expand Down
8 changes: 4 additions & 4 deletions sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/CommonJms.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -143,7 +143,7 @@ Class<? extends ConnectionFactory> 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<String> {

@Override
Expand Down
14 changes: 7 additions & 7 deletions sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down
20 changes: 10 additions & 10 deletions sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down
8 changes: 4 additions & 4 deletions ...io/jms/src/test/java/org/apache/beam/sdk/io/jms/MockNonSerializableConnectionFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
13 changes: 13 additions & 0 deletions sdks/java/io/mqtt/build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.

@damccorm damccorm Jun 30, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see a couple of mentions of making this compatible with Java 11, but then we're forcing Java 17 - do we actually need to force Java 17 or can we just require Java 11?

@Abacn Abacn Jul 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ActiveMQ is a test dependency for AmqpIO, MqttIO, JmsIO. Instead of modifying build.gradle in the former ones, can we just pin to ActiveMQ 6 in JmsIO? This should give a smaller diff and safer as it doesn't do a global upgrade.

@Abacn Abacn Jul 1, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

PS: I'm thinking about we should move our Infra to Java17 (probably Java21, to reduce the frequency of this kind of upgrade) while keep Java11 compatibility via cross-compilation. We already use Java21 to publish Javadoc

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."

Expand Down
Loading