Don't attempt to delete pending ack store unless transactions are enabled
@@ -1011,6 +1011,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
@Override
public CompletableFuture<Void> unsubscribe(String subscriptionName) {
CompletableFuture<Void> unsubscribeFuture = new CompletableFuture<>();
+
+ if (brokerService.pulsar().getConfiguration().isTransactionCoordinatorEnabled()) {
getBrokerService().getManagedLedgerFactory().asyncDelete(TopicName.get(MLPendingAckStore
.getTransactionPendingAckStoreSuffix(topic,
Codec.encode(subscriptionName))).getPersistenceNamingEncoding(),
@@ -1032,6 +1034,9 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
topic, subscriptionName, exception);
}
}, null);
+ } else {
+ asyncDeleteCursor(subscriptionName, unsubscribeFuture);
+ }
return unsubscribeFuture;
}
