[improve] Do not process acks in the Netty thread

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java

@@ -557,22 +557,29 @@ private CompletableFuture<Long> individualAckNormal(CommandAck ack, Map<String,
 
             totalAckCount += ackedCount;
         }
-        subscription.acknowledgeMessage(positionsAcked, AckType.Individual, properties);
-        CompletableFuture<Long> completableFuture = new CompletableFuture<>();
-        completableFuture.complete(totalAckCount);
-        if (isTransactionEnabled() && Subscription.isIndividualAckMode(subType)) {
-            completableFuture.whenComplete((v, e) -> positionsAcked.forEach(position -> {
-                //check if the position can remove from the consumer pending acks.
-                // the bit set is empty in pending ack handle.
-                if (((PositionImpl) position).getAckSet() != null) {
-                    if (((PersistentSubscription) subscription)
-                            .checkIsCanDeleteConsumerPendingAck((PositionImpl) position)) {
-                        removePendingAcks((PositionImpl) position);
+
+        // we use acknowledgeMessageAsync because we don't want to perform the
+        // flush of the cursor (that may take much time in case of a long list of individuallyDeletedMessages)
+        // in the Netty eventloop thread
+        long totalCount = totalAckCount;
+        return subscription.acknowledgeMessageAsync(positionsAcked, AckType.Individual, properties)
+                .thenCompose(___ -> {
+            CompletableFuture<Long> completableFuture = new CompletableFuture<>();
+            completableFuture.complete(totalCount);
+            if (isTransactionEnabled() && Subscription.isIndividualAckMode(subType)) {
+                completableFuture.whenComplete((v, e) -> positionsAcked.forEach(position -> {
+                    //check if the position can remove from the consumer pending acks.
+                    // the bit set is empty in pending ack handle.
+                    if (((PositionImpl) position).getAckSet() != null) {
+                        if (((PersistentSubscription) subscription)
+                                .checkIsCanDeleteConsumerPendingAck((PositionImpl) position)) {
+                            removePendingAcks((PositionImpl) position);
+                        }
                     }
-                }
-            }));
-        }
-        return completableFuture;
+                }));
+            }
+            return completableFuture;
+        });
     }
 
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java

@@ -51,6 +51,12 @@ default void removeConsumer(Consumer consumer) throws BrokerServiceException {
 
     void acknowledgeMessage(List<Position> positions, AckType ackType, Map<String, Long> properties);
 
+    default CompletableFuture<Void> acknowledgeMessageAsync(List<Position> positions,
+                                                            AckType ackType, Map<String, Long> properties) {
+        acknowledgeMessage(positions, ackType, properties);
+        return CompletableFuture.completedFuture(null);
+    }
+
     String getTopicName();
 
     boolean isReplicated();

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

@@ -367,6 +367,14 @@ public void consumerFlow(Consumer consumer, int additionalNumberOfMessages) {
         dispatcher.consumerFlow(consumer, additionalNumberOfMessages);
     }
 
+    @Override
+    public CompletableFuture<Void> acknowledgeMessageAsync(List<Position> positions,
+                                                           AckType ackType, Map<String, Long> properties) {
+        return CompletableFuture.runAsync(() -> {
+                acknowledgeMessage(positions, ackType, properties);
+                }, topic.getBrokerService().getTopicOrderedExecutor().chooseThread(cursor.getName()));
+    }
+
     @Override
     public void acknowledgeMessage(List<Position> positions, AckType ackType, Map<String, Long> properties) {
         cursor.updateLastActive();