[fix][client] Fix NPE when acknowledging multiple messages

pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerAckListTest.java

@@ -111,4 +111,43 @@ private void sendMessagesAsyncAndWait(Producer<String> producer, int messages) t
         latch.await();
     }
 
+    @Test(timeOut = 30000)
+    public void testAckMessageInAnotherTopic() throws Exception {
+        final String[] topics = {
+                "persistent://my-property/my-ns/test-ack-message-in-other-topic1" + UUID.randomUUID(),
+                "persistent://my-property/my-ns/test-ack-message-in-other-topic2" + UUID.randomUUID(),
+                "persistent://my-property/my-ns/test-ack-message-in-other-topic3" + UUID.randomUUID()
+        };
+        @Cleanup final Consumer<String> allTopicsConsumer = pulsarClient.newConsumer(Schema.STRING)
+                .topic(topics)
+                .subscriptionName("sub1")
+                .subscribe();
+        Consumer<String> partialTopicsConsumer = pulsarClient.newConsumer(Schema.STRING)
+                .topic(topics[0], topics[1])
+                .subscriptionName("sub2")
+                .subscribe();
+        for (int i = 0; i < topics.length; i++) {
+            final Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+                    .topic(topics[i])
+                    .create();
+            producer.send("msg-" + i);
+            producer.close();
+        }
+        final List<MessageId> messageIdList = new ArrayList<>();
+        for (int i = 0; i < topics.length; i++) {
+            messageIdList.add(allTopicsConsumer.receive().getMessageId());
+        }
+        try {
+            partialTopicsConsumer.acknowledge(messageIdList);
+            Assert.fail();
+        } catch (PulsarClientException.NotConnectedException ignored) {
+        }
+        partialTopicsConsumer.close();
+        partialTopicsConsumer = pulsarClient.newConsumer(Schema.STRING).topic(topics[0])
+                .subscriptionName("sub2").subscribe();
+        pulsarClient.newProducer(Schema.STRING).topic(topics[0]).create().send("done");
+        final Message<String> msg = partialTopicsConsumer.receive();
+        Assert.assertEquals(msg.getValue(), "msg-0");
+        partialTopicsConsumer.close();
+    }
 }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java

@@ -32,6 +32,7 @@
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.IdentityHashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -494,8 +495,16 @@ protected CompletableFuture<Void> doAcknowledge(List<MessageId> messageIdList,
                 topicToMessageIdMap.get(topicMessageId.getOwnerTopic())
                         .add(MessageIdImpl.convertToMessageIdImpl(topicMessageId));
             }
-            topicToMessageIdMap.forEach((topicPartitionName, messageIds) -> {
-                ConsumerImpl<T> consumer = consumers.get(topicPartitionName);
+            final Map<ConsumerImpl<T>, List<MessageId>> consumerToMessageIds = new IdentityHashMap<>();
+            for (Map.Entry<String, List<MessageId>> entry : topicToMessageIdMap.entrySet()) {
+                ConsumerImpl<T> consumer = consumers.get(entry.getKey());
+                if (consumer == null) {
+                    return FutureUtil.failedFuture(new PulsarClientException.NotConnectedException());
+                }
+                // Trigger the acknowledgment later to avoid sending partial acknowledgments
+                consumerToMessageIds.put(consumer, entry.getValue());
+            }
+            consumerToMessageIds.forEach((consumer, messageIds) -> {
                 resultFutures.add(consumer.doAcknowledgeWithTxn(messageIds, ackType, properties, txn)
                         .thenAccept((res) -> messageIdList.forEach(unAckedMessageTracker::remove)));
             });