sijie commented on a change in pull request #2575: Issue #2574: Timeout message
not get redeliver in TopicsConsumer when use message listener
URL: https://github.com/apache/incubator-pulsar/pull/2575#discussion_r217436845
##########
File path:
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java
##########
@@ -615,4 +616,61 @@ public void testTopicsNameSubscribeWithBuilderFail()
throws Exception {
}
}
+ /**
+ * Test Listener
+ */
+ @Test(timeOut = 30000)
+ public void testMultiTopicsMessageListener() throws Exception {
+ String key = "MultiTopicsMessageListenerTest";
+ final String subscriptionName = "my-ex-subscription-" + key;
+ final String messagePredicate = "my-message-" + key + "-";
+ final int totalMessages = 6;
+
+ // set latch larger than totalMessages, so timeout message get resend
+ CountDownLatch latch = new CountDownLatch(totalMessages * 3);
+
+ final String topicName1 = "persistent://prop/use/ns-abc/topic-1-" +
key;
+ List<String> topicNames = Lists.newArrayList(topicName1);
+
+ admin.tenants().createTenant("prop", new TenantInfo());
+ admin.topics().createPartitionedTopic(topicName1, 2);
+
+ // 1. producer connect
+ Producer<byte[]> producer1 =
pulsarClient.newProducer().topic(topicName1)
+ .enableBatching(false)
+ .messageRoutingMode(MessageRoutingMode.SinglePartition)
+ .create();
+
+ // 2. Create consumer, set not ack in message listener, so time-out
message will resend
+ Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topics(topicNames)
+ .subscriptionName(subscriptionName)
+ .subscriptionType(SubscriptionType.Shared)
+ .ackTimeout(1000, TimeUnit.MILLISECONDS)
+ .receiverQueueSize(4)
+ .messageListener((c1, msg) -> {
+ assertNotNull(msg, "Message cannot be null");
+ String receivedMessage = new String(msg.getData());
+ latch.countDown();
+
+ log.info("Received message [{}] in the listener, latch: {}",
+ receivedMessage, latch.getCount());
+ // since not acked, it should retry another time
Review comment:
nit: we can remove line 658 and 659
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
