rdhabalia commented on code in PR #23345:
URL: https://github.com/apache/pulsar/pull/23345#discussion_r1776429366
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageRedeliveryTest.java:
##########
@@ -539,4 +542,45 @@ public void
testMultiConsumerBatchRedeliveryAddEpoch(boolean enableBatch) throws
// can't receive message again
assertEquals(consumer.batchReceive().size(), 0);
}
+
+ /**
+ * This test validates that failover consumer with message listener keeps
receiving messages even after filtering
+ * out epoch messages.
+ * @throws Exception
+ */
+ @Test
+ public void testRedeliveryWithListenerAddEpoch() throws Exception {
+ final String topic = "testRedeliveryListenerAddEpoch";
+ final String subName = "my-sub";
+ int totalMessages = 100;
+
+ Set<MessageId> ids = new HashSet<>();
+ CountDownLatch latch = new CountDownLatch(totalMessages);
+ MessageListener<String> msgListener = (Consumer<String> consumer,
Message<String> msg) -> {
+ String id = msg.getMessageId().toString();
+ consumer.acknowledgeCumulativeAsync(msg);
+ // increase epoch to make next message invalid and filter out for
the next redelivery
+ long epoch = ((ConsumerBase) consumer).getConsumerEpoch() + 1;
+ ((ConsumerBase) consumer).setConsumerEpoch(epoch);
+ if (ids.add(msg.getMessageId())) {
+ // count unique messages
+ latch.countDown();
+ }
+ consumer.redeliverUnacknowledgedMessages();
+ };
+ ConsumerBase<String> consumer = ((ConsumerBase<String>)
pulsarClient.newConsumer(Schema.STRING).topic(topic)
+
.subscriptionName(subName).messageListener(msgListener).subscriptionType(SubscriptionType.Failover)
+ .receiverQueueSize(totalMessages / 10).subscribe());
+
+ Producer<String> producer =
pulsarClient.newProducer(Schema.STRING).topic(topic).enableBatching(false)
+ .create();
+
+ for (int i = 0; i < totalMessages; i++) {
+ producer.sendAsync("test" + i);
+ }
+ producer.flush();
+
+ latch.await(60, TimeUnit.SECONDS);
Review Comment:
added `@Cleanup` annotation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
