Update Pulsar client to 2.8.1 and imported third party libs like Netty

datastax · Nov 10, 2021 · 25e60f3 · 25e60f3
1 parent 6fc9641
commit 25e60f3
Show file tree

Hide file tree

Showing 7 changed files with 20 additions and 25 deletions.
diff --git a/pom.xml b/pom.xml
@@ -48,7 +48,7 @@
     <maven.compiler.source>8</maven.compiler.source>
     <maven.compiler.target>8</maven.compiler.target>
     <jms.version>2.0.3</jms.version>
-    <pulsar.version>2.8.0</pulsar.version>
+    <pulsar.version>2.8.1</pulsar.version>
     <activemq.version>5.16.1</activemq.version>
     <hawtbuf.version>1.11</hawtbuf.version>
     <curator.version>5.1.0</curator.version>

diff --git a/pulsar-jms-all/src/license/override-THIRD-PARTY.properties b/pulsar-jms-all/src/license/override-THIRD-PARTY.properties
@@ -15,4 +15,4 @@
 #
 
 net.jcip--jcip-annotations--1.0==Apache License, Version 2.0
-io.netty--netty-tcnative-boringssl-static--2.0.38.Final==Apache License, Version 2.0
+io.netty--netty-tcnative-boringssl-static--2.0.40.Final==Apache License, Version 2.0
diff --git a/pulsar-jms-integration-tests/src/test/java/com/datastax/oss/pulsar/jms/tests/DockerTest.java b/pulsar-jms-integration-tests/src/test/java/com/datastax/oss/pulsar/jms/tests/DockerTest.java
@@ -36,13 +36,13 @@ public void testPulsar272() throws Exception {
   }
 
   @Test
-  public void testPulsar280() throws Exception {
-    test("apachepulsar/pulsar:2.8.0", false);
+  public void testPulsar281() throws Exception {
+    test("apachepulsar/pulsar:2.8.1", false);
   }
 
   @Test
-  public void testPulsar280Transactions() throws Exception {
-    test("apachepulsar/pulsar:2.8.0", true);
+  public void testPulsar281Transactions() throws Exception {
+    test("apachepulsar/pulsar:2.8.1", true);
   }
 
   private void test(String image, boolean transactions) throws Exception {

diff --git a/pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PulsarConnectionFactory.java b/pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PulsarConnectionFactory.java
@@ -29,21 +29,8 @@
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Stream;
-import javax.jms.ConnectionFactory;
-import javax.jms.Destination;
+import javax.jms.*;
 import javax.jms.IllegalStateException;
-import javax.jms.InvalidClientIDException;
-import javax.jms.JMSContext;
-import javax.jms.JMSException;
-import javax.jms.JMSRuntimeException;
-import javax.jms.JMSSecurityException;
-import javax.jms.JMSSecurityRuntimeException;
-import javax.jms.Queue;
-import javax.jms.QueueConnection;
-import javax.jms.QueueConnectionFactory;
-import javax.jms.Topic;
-import javax.jms.TopicConnection;
-import javax.jms.TopicConnectionFactory;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminException;
@@ -893,6 +880,8 @@ public Consumer<byte[]> createConsumer(
               .newConsumer()
               // these properties can be overridden by the configuration
               .negativeAckRedeliveryDelay(1, TimeUnit.SECONDS)
+              // always set this for Transactions
+              .isAckReceiptEnabled(sessionMode == Session.SESSION_TRANSACTED)
               .loadConf(getConsumerConfiguration())
               // these properties cannot be overwritten by the configuration
               .subscriptionInitialPosition(initialPosition)
@@ -973,8 +962,14 @@ public boolean deleteSubscription(PulsarDestination destination, String name)
         List<String> allTopics = pulsarAdmin.topics().getList(systemNamespace);
         for (String topic : allTopics) {
           log.info("Scanning topic {}", topic);
-          List<String> subscriptions = pulsarAdmin.topics().getSubscriptions(topic);
-          log.info("Subscriptions {}", subscriptions);
+          List<String> subscriptions;
+          try {
+            subscriptions = pulsarAdmin.topics().getSubscriptions(topic);
+            log.info("Subscriptions {}", subscriptions);
+          } catch (PulsarAdminException.NotFoundException notFound) {
+            log.error("Skipping topic {}", topic);
+            subscriptions = Collections.emptyList();
+          }
           for (String subscription : subscriptions) {
             log.info("Found subscription {} ", subscription);
             if (subscription.equals(name)) {

diff --git a/resource-adapter-tests/src/test/java/com/datastax/oss/pulsar/jms/tests/PulsarContainer.java b/resource-adapter-tests/src/test/java/com/datastax/oss/pulsar/jms/tests/PulsarContainer.java
@@ -34,7 +34,7 @@ public PulsarContainer(Network network) {
   public void start() throws Exception {
     CountDownLatch pulsarReady = new CountDownLatch(1);
     pulsarContainer =
-        new GenericContainer<>("apachepulsar/pulsar:2.8.0")
+        new GenericContainer<>("apachepulsar/pulsar:2.8.1")
             .withNetwork(network)
             .withNetworkAliases("pulsar")
             .withCommand(

diff --git a/resource-adapter/src/license/override-THIRD-PARTY.properties b/resource-adapter/src/license/override-THIRD-PARTY.properties
@@ -15,4 +15,4 @@
 #
 
 net.jcip--jcip-annotations--1.0==Apache License, Version 2.0
-io.netty--netty-tcnative-boringssl-static--2.0.38.Final==Apache License, Version 2.0
+io.netty--netty-tcnative-boringssl-static--2.0.40.Final==Apache License, Version 2.0
diff --git a/tck-executor/start_pulsar.sh b/tck-executor/start_pulsar.sh
@@ -2,7 +2,7 @@
 
 set -x -e
 
-IMAGENAME=apachepulsar/pulsar:2.8.0
+IMAGENAME=apachepulsar/pulsar:2.8.1
 
 HERE=$(dirname $0)
 HERE=$(realpath "$HERE")