Fix NPE when acknowledge messages at the broker side
@@ -361,8 +361,10 @@ public class PersistentSubscription implements Subscription {
cursor.asyncDelete(positions, deleteCallback, positions);
}
+ if(dispatcher != null){
dispatcher.getRedeliveryTracker().removeBatch(positions);
}
+ }
if (!cursor.getMarkDeletedPosition().equals(previousMarkDeletePosition)) {
// Mark delete position advance
@@ -379,12 +381,16 @@ public class PersistentSubscription implements Subscription {
if (topic.getManagedLedger().isTerminated() && cursor.getNumberOfEntriesInBacklog(false) == 0) {
// Notify all consumer that the end of topic was reached
+ if(dispatcher != null){
dispatcher.getConsumers().forEach(Consumer::reachedEndOfTopic);
}
+ }
// Signal the dispatchers to give chance to take extra actions
+ if(dispatcher != null){
dispatcher.acknowledgementWasProcessed();
}
+ }
/**
* Acknowledge message(s) for an ongoing transaction.
