Skip to content

Commit a1f3ae2

Browse files
jbonofrejbonofre
committed
Update JmsIO to ActiveMQ 6.2.5 and jakarta.jms
Bump activemq to 6.2.5 and qpid-jms-client to 2.10.0, both of which target jakarta.jms. Replace the geronimo-jms_2.0_spec dependency with jakarta.jms-api 3.1.0 and migrate all javax.jms imports and javadoc references in JmsIO main and test sources to jakarta.jms.
1 parent eaba570 commit a1f3ae2

15 files changed

Lines changed: 89 additions & 68 deletions

File tree

‎.github/workflows/beam_PreCommit_Java_Jms_IO_Direct.yml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,8 @@ jobs:
8585
github_job: ${{ matrix.job_name }} (${{ matrix.job_phrase }})
8686
- name: Setup environment
8787
uses: ./.github/actions/setup-environment-action
88+
with:
89+
java-version: 17
8890
- name: run Jms IO build script
8991
uses: ./.github/actions/gradle-command-self-hosted-action
9092
with:

‎CHANGES.md‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,9 @@
8282
* (Python) Typehints of dataclass fields are honored during type inferences. To restore the behavior of fallback-to-any,
8383
use pipeline option `--exclude_infer_dataclass_field_type` ([#38797](https://github.com/apache/beam/issues/38797)).
8484
However fixing forward is recommended.
85+
* (Java) JmsIO migrated to `jakarta.jms` (JMS 3.1) and ActiveMQ 6.2.5. User code implementing `JmsIO.MessageMapper`,
86+
`valueMapper`, `topicNameMapper`, or providing a `ConnectionFactory` must update imports from `javax.jms.*` to
87+
`jakarta.jms.*`. The module now requires Java 17 at runtime ([#38729](https://github.com/apache/beam/issues/38729)).
8588

8689
## Deprecations
8790

‎buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -598,6 +598,7 @@ class BeamModulePlugin implements Plugin<Project> {
598598
// There are a few versions are determined by the BOMs by running scripts/tools/bomupgrader.py
599599
// marked as [bomupgrader]. See the documentation of that script for detail.
600600
def activemq_version = "5.19.2"
601+
def activemq6_version = "6.2.5"
601602
def autovalue_version = "1.9"
602603
def autoservice_version = "1.0.1"
603604
def aws_java_sdk2_version = "2.20.162"
@@ -639,7 +640,7 @@ class BeamModulePlugin implements Plugin<Project> {
639640
def postgres_version = "42.6.2"
640641
// [bomupgrader] determined by: com.google.protobuf:protobuf-java, consistent with: google_cloud_platform_libraries_bom
641642
def protobuf_version = "4.33.2"
642-
def qpid_jms_client_version = "0.61.0"
643+
def qpid_jms_client_version = "2.10.0"
643644
def quickcheck_version = "1.0"
644645
def sbe_tool_version = "1.25.1"
645646
def singlestore_jdbc_version = "1.1.4"
@@ -678,6 +679,13 @@ class BeamModulePlugin implements Plugin<Project> {
678679
activemq_junit : "org.apache.activemq.tooling:activemq-junit:$activemq_version",
679680
activemq_kahadb_store : "org.apache.activemq:activemq-kahadb-store:$activemq_version",
680681
activemq_mqtt : "org.apache.activemq:activemq-mqtt:$activemq_version",
682+
activemq6_amqp : "org.apache.activemq:activemq-amqp:$activemq6_version",
683+
activemq6_broker : "org.apache.activemq:activemq-broker:$activemq6_version",
684+
activemq6_client : "org.apache.activemq:activemq-client:$activemq6_version",
685+
activemq6_jaas : "org.apache.activemq:activemq-jaas:$activemq6_version",
686+
activemq6_junit : "org.apache.activemq.tooling:activemq-junit:$activemq6_version",
687+
activemq6_kahadb_store : "org.apache.activemq:activemq-kahadb-store:$activemq6_version",
688+
activemq6_mqtt : "org.apache.activemq:activemq-mqtt:$activemq6_version",
681689
aircompressor : "io.airlift:aircompressor:2.0.3",
682690
args4j : "args4j:args4j:2.33",
683691
auto_value_annotations : "com.google.auto.value:auto-value-annotations:$autovalue_version",

‎sdks/java/io/amqp/build.gradle‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,10 @@
1717
*/
1818

1919
plugins { id 'org.apache.beam.module' }
20-
applyJavaNature( automaticModuleName: 'org.apache.beam.sdk.io.amqp')
20+
applyJavaNature(
21+
automaticModuleName: 'org.apache.beam.sdk.io.amqp',
22+
requireJavaVersion: JavaVersion.VERSION_17,
23+
)
2124

2225
description = "Apache Beam :: SDKs :: Java :: IO :: AMQP"
2326
ext.summary = "IO to read and write using AMQP 1.0 protocol (http://www.amqp.org)."
@@ -29,11 +32,11 @@ dependencies {
2932
implementation "org.apache.qpid:proton-j:0.16.0"
3033
testImplementation library.java.slf4j_api
3134
testImplementation library.java.junit
32-
testImplementation library.java.activemq_broker
33-
testImplementation(library.java.activemq_amqp) {
35+
testImplementation library.java.activemq6_broker
36+
testImplementation(library.java.activemq6_amqp) {
3437
exclude group: 'org.apache.qpid', module: 'proton-j'
3538
}
36-
testImplementation library.java.activemq_junit
39+
testImplementation library.java.activemq6_junit
3740
testImplementation library.java.hamcrest
3841
testRuntimeOnly library.java.slf4j_jdk14
3942
testRuntimeOnly project(path: ":runners:direct-java", configuration: "shadow")

‎sdks/java/io/jms/build.gradle‎

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
plugins { id 'org.apache.beam.module' }
2020
applyJavaNature(
2121
automaticModuleName: 'org.apache.beam.sdk.io.jms',
22+
requireJavaVersion: JavaVersion.VERSION_17,
2223
)
2324
provideIntegrationTestingDependencies()
2425
enableJavaPerformanceTesting()
@@ -32,12 +33,12 @@ dependencies {
3233
implementation project(path: ":sdks:java:core", configuration: "shadow")
3334
implementation library.java.slf4j_api
3435
implementation library.java.joda_time
35-
implementation "org.apache.geronimo.specs:geronimo-jms_2.0_spec:1.0-alpha-2"
36-
testImplementation library.java.activemq_amqp
37-
testImplementation library.java.activemq_broker
38-
testImplementation library.java.activemq_jaas
39-
testImplementation library.java.activemq_kahadb_store
40-
testImplementation library.java.activemq_client
36+
implementation "jakarta.jms:jakarta.jms-api:3.1.0"
37+
testImplementation library.java.activemq6_amqp
38+
testImplementation library.java.activemq6_broker
39+
testImplementation library.java.activemq6_jaas
40+
testImplementation library.java.activemq6_kahadb_store
41+
testImplementation library.java.activemq6_client
4142
testImplementation library.java.hamcrest
4243
testImplementation library.java.junit
4344
testImplementation library.java.mockito_core

‎sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,14 +17,14 @@
1717
*/
1818
package org.apache.beam.sdk.io.jms;
1919

20+
import jakarta.jms.JMSException;
21+
import jakarta.jms.Message;
22+
import jakarta.jms.MessageConsumer;
23+
import jakarta.jms.Session;
2024
import java.io.IOException;
2125
import java.io.Serializable;
2226
import java.util.Objects;
2327
import java.util.concurrent.locks.ReentrantReadWriteLock;
24-
import javax.jms.JMSException;
25-
import javax.jms.Message;
26-
import javax.jms.MessageConsumer;
27-
import javax.jms.Session;
2828
import org.apache.beam.sdk.io.UnboundedSource;
2929
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
3030
import org.checkerframework.checker.nullness.qual.Nullable;

‎sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java‎

Lines changed: 17 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,15 @@
2222

2323
import com.google.auto.value.AutoValue;
2424
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
25+
import jakarta.jms.Connection;
26+
import jakarta.jms.ConnectionFactory;
27+
import jakarta.jms.Destination;
28+
import jakarta.jms.JMSException;
29+
import jakarta.jms.Message;
30+
import jakarta.jms.MessageConsumer;
31+
import jakarta.jms.MessageProducer;
32+
import jakarta.jms.Session;
33+
import jakarta.jms.TextMessage;
2534
import java.io.IOException;
2635
import java.io.Serializable;
2736
import java.nio.charset.StandardCharsets;
@@ -36,15 +45,6 @@
3645
import java.util.concurrent.ScheduledExecutorService;
3746
import java.util.concurrent.TimeUnit;
3847
import java.util.stream.Stream;
39-
import javax.jms.Connection;
40-
import javax.jms.ConnectionFactory;
41-
import javax.jms.Destination;
42-
import javax.jms.JMSException;
43-
import javax.jms.Message;
44-
import javax.jms.MessageConsumer;
45-
import javax.jms.MessageProducer;
46-
import javax.jms.Session;
47-
import javax.jms.TextMessage;
4848
import org.apache.beam.sdk.coders.Coder;
4949
import org.apache.beam.sdk.coders.SerializableCoder;
5050
import org.apache.beam.sdk.io.Read.Unbounded;
@@ -86,9 +86,9 @@
8686
*
8787
* <p>JmsIO source returns unbounded collection of JMS records as {@code PCollection<JmsRecord>}. A
8888
* {@link JmsRecord} includes JMS headers and properties, along with the JMS {@link
89-
* javax.jms.TextMessage} payload.
89+
* jakarta.jms.TextMessage} payload.
9090
*
91-
* <p>To configure a JMS source, you have to provide a {@link javax.jms.ConnectionFactory} and the
91+
* <p>To configure a JMS source, you have to provide a {@link jakarta.jms.ConnectionFactory} and the
9292
* destination (queue or topic) where to consume. The following example illustrates various options
9393
* for configuring the source:
9494
*
@@ -102,8 +102,8 @@
102102
*
103103
* }</pre>
104104
*
105-
* <p>It is possible to read any type of JMS {@link javax.jms.Message} into a custom POJO using the
106-
* following configuration:
105+
* <p>It is possible to read any type of JMS {@link jakarta.jms.Message} into a custom POJO using
106+
* the following configuration:
107107
*
108108
* <pre>{@code
109109
* pipeline.apply(JmsIO.<T>readMessage()
@@ -121,8 +121,8 @@
121121
* <h3>Writing to a JMS destination</h3>
122122
*
123123
* <p>JmsIO sink supports writing text messages to a JMS destination on a broker. To configure a JMS
124-
* sink, you must specify a {@link javax.jms.ConnectionFactory} and a {@link javax.jms.Destination}
125-
* name. For instance:
124+
* sink, you must specify a {@link jakarta.jms.ConnectionFactory} and a {@link
125+
* jakarta.jms.Destination} name. For instance:
126126
*
127127
* <pre>{@code
128128
* pipeline
@@ -1053,7 +1053,7 @@ public Write<EventT> withTopicNameMapper(SerializableFunction<EventT, String> to
10531053
}
10541054

10551055
/**
1056-
* Map the {@code EventT} object to a {@link javax.jms.Message}.
1056+
* Map the {@code EventT} object to a {@link jakarta.jms.Message}.
10571057
*
10581058
* <p>For instance:
10591059
*
@@ -1075,7 +1075,7 @@ public Write<EventT> withTopicNameMapper(SerializableFunction<EventT, String> to
10751075
* .apply(JmsIO.write().withValueMapper(valueNapper)
10761076
* }</pre>
10771077
*
1078-
* @param valueMapper The function returning the {@link javax.jms.Message}
1078+
* @param valueMapper The function returning the {@link jakarta.jms.Message}
10791079
* @return The corresponding {@link JmsIO.Write}.
10801080
*/
10811081
public Write<EventT> withValueMapper(

‎sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsRecord.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,10 +18,10 @@
1818
package org.apache.beam.sdk.io.jms;
1919

2020
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
21+
import jakarta.jms.Destination;
2122
import java.io.Serializable;
2223
import java.util.Map;
2324
import java.util.Objects;
24-
import javax.jms.Destination;
2525
import org.checkerframework.checker.nullness.qual.Nullable;
2626

2727
/**

‎sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/TextMessageMapper.java‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -17,15 +17,15 @@
1717
*/
1818
package org.apache.beam.sdk.io.jms;
1919

20-
import javax.jms.JMSException;
21-
import javax.jms.Message;
22-
import javax.jms.Session;
23-
import javax.jms.TextMessage;
20+
import jakarta.jms.JMSException;
21+
import jakarta.jms.Message;
22+
import jakarta.jms.Session;
23+
import jakarta.jms.TextMessage;
2424
import org.apache.beam.sdk.transforms.SerializableBiFunction;
2525

2626
/**
27-
* The TextMessageMapper takes a {@link String} value, a {@link javax.jms.Session} and returns a
28-
* {@link javax.jms.TextMessage}.
27+
* The TextMessageMapper takes a {@link String} value, a {@link jakarta.jms.Session} and returns a
28+
* {@link jakarta.jms.TextMessage}.
2929
*/
3030
public class TextMessageMapper implements SerializableBiFunction<String, Session, Message> {
3131

‎sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/CommonJms.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,15 +17,15 @@
1717
*/
1818
package org.apache.beam.sdk.io.jms;
1919

20+
import jakarta.jms.BytesMessage;
21+
import jakarta.jms.ConnectionFactory;
22+
import jakarta.jms.Message;
2023
import java.io.Serializable;
2124
import java.lang.reflect.InvocationTargetException;
2225
import java.nio.charset.StandardCharsets;
2326
import java.util.ArrayList;
2427
import java.util.List;
2528
import java.util.function.Supplier;
26-
import javax.jms.BytesMessage;
27-
import javax.jms.ConnectionFactory;
28-
import javax.jms.Message;
2929
import org.apache.activemq.ActiveMQConnectionFactory;
3030
import org.apache.activemq.broker.BrokerPlugin;
3131
import org.apache.activemq.broker.BrokerService;
@@ -143,7 +143,7 @@ Class<? extends ConnectionFactory> getConnectionFactoryClass() {
143143
return this.connectionFactoryClass;
144144
}
145145

146-
/** A test class that maps a {@link javax.jms.BytesMessage} into a {@link String}. */
146+
/** A test class that maps a {@link jakarta.jms.BytesMessage} into a {@link String}. */
147147
public static class BytesMessageToStringMessageMapper implements JmsIO.MessageMapper<String> {
148148

149149
@Override

0 commit comments

Comments
 (0)