Ignore system topics (workaround for apache/pulsar#12727)

datastax · Nov 10, 2021 · 4c4b6dc · 4c4b6dc
1 parent 25e60f3
commit 4c4b6dc
Showing 1 changed file with 8 additions and 2 deletions.
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
@@ -55,6 +55,9 @@
 @Slf4j
 public class PulsarConnectionFactory
     implements ConnectionFactory, QueueConnectionFactory, TopicConnectionFactory, AutoCloseable {
+
+  private static final String PENDING_ACK_STORE_SUFFIX = "__transaction_pending_ack";
+
   private static final Set<String> clientIdentifiers = new ConcurrentSkipListSet<>();
 
   private final Map<String, Producer<byte[]>> producers = new ConcurrentHashMap<>();
@@ -880,8 +883,6 @@ 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)
@@ -961,6 +962,11 @@ public boolean deleteSubscription(PulsarDestination destination, String name)
         // required for TCK, scan for all subscriptions
         List<String> allTopics = pulsarAdmin.topics().getList(systemNamespace);
         for (String topic : allTopics) {
+          if (topic.endsWith(PENDING_ACK_STORE_SUFFIX)) {
+            // skip Transaction related system topics
+            log.info("Ignoring system topic {}", topic);
+            continue;
+          }
           log.info("Scanning topic {}", topic);
           List<String> subscriptions;
           try {