diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerAckListTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerAckListTest.java index d01e5b764a7a1..baf0000be6e4f 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerAckListTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ConsumerAckListTest.java @@ -111,4 +111,43 @@ private void sendMessagesAsyncAndWait(Producer 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 allTopicsConsumer = pulsarClient.newConsumer(Schema.STRING) + .topic(topics) + .subscriptionName("sub1") + .subscribe(); + Consumer partialTopicsConsumer = pulsarClient.newConsumer(Schema.STRING) + .topic(topics[0], topics[1]) + .subscriptionName("sub2") + .subscribe(); + for (int i = 0; i < topics.length; i++) { + final Producer producer = pulsarClient.newProducer(Schema.STRING) + .topic(topics[i]) + .create(); + producer.send("msg-" + i); + producer.close(); + } + final List 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 msg = partialTopicsConsumer.receive(); + Assert.assertEquals(msg.getValue(), "msg-0"); + partialTopicsConsumer.close(); + } } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java index 341a91e97348e..f993304b0780a 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java +++ b/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 doAcknowledge(List messageIdList, topicToMessageIdMap.get(topicMessageId.getOwnerTopic()) .add(MessageIdImpl.convertToMessageIdImpl(topicMessageId)); } - topicToMessageIdMap.forEach((topicPartitionName, messageIds) -> { - ConsumerImpl consumer = consumers.get(topicPartitionName); + final Map, List> consumerToMessageIds = new IdentityHashMap<>(); + for (Map.Entry> entry : topicToMessageIdMap.entrySet()) { + ConsumerImpl 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))); });