Add logic to call the proper delete method based on whether the topic…

… is partitioned or not (#110)
datastax · Aug 2, 2023 · 83a5a4c · 83a5a4c
1 parent 4a5757b
commit 83a5a4c
Showing 1 changed file with 22 additions and 7 deletions.
diff --git a/pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PulsarTemporaryDestination.java b/pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PulsarTemporaryDestination.java
@@ -18,6 +18,7 @@
 import javax.jms.InvalidDestinationException;
 import javax.jms.JMSException;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.common.policies.data.TopicStats;
 
 @Slf4j
@@ -53,16 +54,30 @@ public final void delete() throws JMSException {
         throw new JMSException("Cannot delete a temporary destination with active consumers");
       }
 
-      session
+      if (session
           .getFactory()
           .getPulsarAdmin()
           .topics()
-          .delete(
-              fullQualifiedTopicName,
-              session.getFactory().isForceDeleteTemporaryDestinations(),
-              true);
-    } catch (Exception err) {
-      throw Utils.handleException(err);
+          .getPartitionedTopicList(session.getFactory().getSystemNamespace())
+          .stream()
+          .anyMatch(t -> t.equals(fullQualifiedTopicName))) {
+        session
+            .getFactory()
+            .getPulsarAdmin()
+            .topics()
+            .deletePartitionedTopic(
+                fullQualifiedTopicName, session.getFactory().isForceDeleteTemporaryDestinations());
+      } else {
+        session
+            .getFactory()
+            .getPulsarAdmin()
+            .topics()
+            .delete(
+                fullQualifiedTopicName, session.getFactory().isForceDeleteTemporaryDestinations());
+      }
+
+    } catch (final PulsarAdminException paEx) {
+      Utils.handleException(paEx);
     } finally {
       session.getConnection().removeTemporaryDestination(this);
     }