How to achieve the effect of maximum delay time in retransmission backoff mechanism #18261

startjava · 2022-10-31T03:21:43Z

startjava
Oct 31, 2022

    @Bean
    public Consumer createPulsarConsumer_myTopic2() throws PulsarClientException {
        Consumer consumer = pulsarClient.newConsumer().topic(myTopic2).subscriptionName(myTopic2_subscriptionName).messageListener(myMessageListener1).subscriptionType(SubscriptionType.Shared).ackTimeout(10, TimeUnit.SECONDS).negativeAckRedeliveryBackoff(MultiplierRedeliveryBackoff.builder()
                .minDelayMs(1000)
                .maxDelayMs(60 * 1000)
                .multiplier(2)
                .build()).subscribe();
        return consumer;
    }


@Component
public class MyMessageListener1 implements MessageListener {
    @Override
    public void received(Consumer consumer, Message msg) {
        String messageValue = new String(msg.getData());
        messageValue = messageValue.split("\\_")[0];
        System.out.println(System.currentTimeMillis() + " MyMessageListener1=" + messageValue);
    }
}

    @RequestMapping("sendMessage5")
    public String sendMessage5() throws PulsarClientException {
        String messageValue = "我是消息4";
        producer3.send(messageValue);
        return "成功生产5";
    }

My current effect produces a message, then nack, and then Redelivery every 10s,,

I want to know when is maxDelayMs (60 * 1000)? thank
i want achieve bottom effect：

Redelivery count	Redelivery delay
1	10 + 1 seconds
2	10 + 2 seconds
3	10 + 4 seconds
4	10 + 8 seconds
5	10 + 16 seconds
6	10 + 32 seconds
7	10 + 60 seconds
8	10 + 60 seconds

but now every 10s ，console print info：
1667190123029 MyMessageListener1=我是消息4
2022-10-31 12:22:13.306 INFO 14156 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='b14c6', topic='myTopic2'}] 1 messages will be re-delivered
1667190133442 MyMessageListener1=我是消息4
2022-10-31 12:22:24.323 INFO 14156 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='b14c6', topic='myTopic2'}] 1 messages will be re-delivered
1667190144493 MyMessageListener1=我是消息4
2022-10-31 12:22:35.340 INFO 14156 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='b14c6', topic='myTopic2'}] 1 messages will be re-delivered
1667190155552 MyMessageListener1=我是消息4
2022-10-31 12:22:46.353 INFO 14156 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='b14c6', topic='myTopic2'}] 1 messages will be re-delivered
1667190166406 MyMessageListener1=我是消息4
2022-10-31 12:22:57.369 INFO 14156 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='b14c6', topic='myTopic2'}] 1 messages will be re-delivered
1667190177575 MyMessageListener1=我是消息4
2022-10-31 12:23:01.286 INFO 14156 --- [ulsar-timer-6-1] o.a.p.c.impl.ConsumerStatsRecorderImpl : [myTopic2] [myTopic2_subscriptionName1] [b14c6] Prefetched messages: 0 --- Consume throughput received: 0.10 msgs/s --- 0.00 Mbit/s --- Ack sent rate: 0.00 ack/s --- Failed messages: 0 --- batch messages: 0 ---Failed acks: 0
2022-10-31 12:23:08.385 INFO 14156 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='b14c6', topic='myTopic2'}] 1 messages will be re-delivered
1667190188395 MyMessageListener1=我是消息4
2022-10-31 12:23:19.400 INFO 14156 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='b14c6', topic='myTopic2'}] 1 messages will be re-delivered
1667190199517 MyMessageListener1=我是消息4
2022-10-31 12:23:30.419 INFO 14156 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='b14c6', topic='myTopic2'}] 1 messages will be re-delivered
1667190210430 MyMessageListener1=我是消息4
2022-10-31 12:23:41.434 INFO 14156 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='b14c6', topic='myTopic2'}] 1 messages will be re-delivered
1667190221831 MyMessageListener1=我是消息4

codelipenghui · 2022-10-31T04:13:18Z

codelipenghui
Oct 31, 2022
Collaborator

@startjava Have you tried to use a Shared subscription? By default, Pulsar will use the exclusive subscription. Pulsar will not track the redelivery count for exclusive and failover subscriptions.

2 replies

startjava Oct 31, 2022
Author

@codelipenghui
i use shared subscription.
top code i update .

startjava Oct 31, 2022
Author

every 10s。

startjava · 2022-10-31T04:28:17Z

startjava
Oct 31, 2022
Author

but i run nack operate：
@component
public class MyMessageListener1 implements MessageListener {
@OverRide
public void received(Consumer consumer, Message msg) {
String messageValue = new String(msg.getData());
messageValue = messageValue.split("\_")[0];
System.out.println(System.currentTimeMillis() + " MyMessageListener1=" + messageValue);
consumer.negativeAcknowledge(msg);
}
}

print time info :
1667190365861 MyMessageListener1=我是消息4
1667190366883 MyMessageListener1=我是消息4
1667190369261 MyMessageListener1=我是消息4
1667190373664 MyMessageListener1=我是消息4
2022-10-31 12:26:16.697 INFO 16612 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='7846e', topic='myTopic2'}] 1 messages will be re-delivered
1667190376740 MyMessageListener1=我是消息4
2022-10-31 12:26:27.715 INFO 16612 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='7846e', topic='myTopic2'}] 1 messages will be re-delivered
1667190387881 MyMessageListener1=我是消息4
2022-10-31 12:26:38.733 INFO 16612 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='7846e', topic='myTopic2'}] 1 messages will be re-delivered
1667190399052 MyMessageListener1=我是消息4
2022-10-31 12:26:49.748 INFO 16612 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='7846e', topic='myTopic2'}] 1 messages will be re-delivered
1667190410015 MyMessageListener1=我是消息4
2022-10-31 12:26:58.666 INFO 16612 --- [ulsar-timer-6-1] o.a.p.c.impl.ConsumerStatsRecorderImpl : [myTopic2] [myTopic2_subscriptionName1] [7846e] Prefetched messages: 0 --- Consume throughput received: 0.13 msgs/s --- 0.00 Mbit/s --- Ack sent rate: 0.00 ack/s --- Failed messages: 0 --- batch messages: 0 ---Failed acks: 0
2022-10-31 12:27:00.764 INFO 16612 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='7846e', topic='myTopic2'}] 1 messages will be re-delivered
1667190420815 MyMessageListener1=我是消息4
2022-10-31 12:27:11.780 INFO 16612 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='7846e', topic='myTopic2'}] 1 messages will be re-delivered
1667190431887 MyMessageListener1=我是消息4
2022-10-31 12:27:22.794 INFO 16612 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='7846e', topic='myTopic2'}] 1 messages will be re-delivered
1667190442939 MyMessageListener1=我是消息4
2022-10-31 12:27:33.809 INFO 16612 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='7846e', topic='myTopic2'}] 1 messages will be re-delivered
1667190453890 MyMessageListener1=我是消息4
2022-10-31 12:27:44.825 INFO 16612 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='7846e', topic='myTopic2'}] 1 messages will be re-delivered
1667190465176 MyMessageListener1=我是消息4
2022-10-31 12:27:55.838 INFO 16612 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='7846e', topic='myTopic2'}] 1 messages will be re-delivered
1667190476165 MyMessageListener1=我是消息4
2022-10-31 12:27:58.667 INFO 16612 --- [ulsar-timer-6-1] o.a.p.c.impl.ConsumerStatsRecorderImpl : [myTopic2] [myTopic2_subscriptionName1] [7846e] Prefetched messages: 0 --- Consume throughput received: 0.10 msgs/s --- 0.00 Mbit/s --- Ack sent rate: 0.00 ack/s --- Failed messages: 0 --- batch messages: 0 ---Failed acks: 0
2022-10-31 12:28:06.856 INFO 16612 --- [ulsar-timer-6-1] o.a.p.client.impl.UnAckedMessageTracker : [ConsumerBase{subscription='myTopic2_subscriptionName1', consumerName='7846e', topic='myTopic2'}] 1 messages will be re-delivered
1667190486986 MyMessageListener1=我是消息4

1 reply

startjava Oct 31, 2022
Author

last effect every 10s redelivery

codelipenghui · 2022-10-31T06:39:16Z

codelipenghui
Oct 31, 2022
Collaborator

@startjava I have pushed out a PR to add the redelivery delay validation #18264

0 replies

startjava · 2022-10-31T11:19:14Z

startjava
Oct 31, 2022
Author

@codelipenghui

    @RequestMapping("sendMessage5")
    public String sendMessage5() throws PulsarClientException {
        String messageValue = "我是消息";
        producer3.send(messageValue);
        return "成功生产5";
    }

@Component
public class MyMessageListener2 implements MessageListener {
    @Override
    public void received(Consumer consumer, Message msg) {
        String messageValue = new String(msg.getData());
        messageValue = messageValue.split("\\_")[0];
        System.out.println("timeList.add(\"" + System.currentTimeMillis() + "\");");
        consumer.negativeAcknowledge(msg);
    }
}

    @Bean
    public Consumer createPulsarConsumer_myTopic2() throws PulsarClientException {
        Consumer consumer = pulsarClient.newConsumer().topic(myTopic2).subscriptionName(myTopic2_subscriptionName).messageListener(myMessageListener2).subscriptionType(SubscriptionType.Shared).negativeAckRedeliveryBackoff(MultiplierRedeliveryBackoff.builder()
                .minDelayMs(1000)
                .maxDelayMs(60 * 1000)
                .build()).subscribe();
        return consumer;
    }

run result the intervals are as follows:
1080
2389
4297
8500
16284
32255
60067

but doc url：
https://pulsar.apache.org/docs/concepts-messaging/#negative-acknowledgement
say +10s，but i run result no +10s，
i code wrong or doc wrong ??
thank you !

3 replies

codelipenghui Oct 31, 2022
Collaborator

It should be documentation issue
The latest documentation works fine https://pulsar.apache.org/docs/next/concepts-messaging/#negative-acknowledgment

/cc @Anonymitaet @momo-jun

startjava Oct 31, 2022
Author

@codelipenghui hahaha , code run result and new doc same ! my code is right thank you !

momo-jun Nov 1, 2022
Collaborator

@codelipenghui @startjava Thanks for pointing this out. The doc update only covered the next version. I will get it fixed.

How to achieve the effect of maximum delay time in retransmission backoff mechanism #18261

Uh oh!

Uh oh!

startjava Oct 31, 2022

Replies: 4 comments · 6 replies

Uh oh!

codelipenghui Oct 31, 2022 Collaborator

Uh oh!

startjava Oct 31, 2022 Author

Uh oh!

startjava Oct 31, 2022 Author

Uh oh!

startjava Oct 31, 2022 Author

Uh oh!

startjava Oct 31, 2022 Author

Uh oh!

codelipenghui Oct 31, 2022 Collaborator

Uh oh!

startjava Oct 31, 2022 Author

Uh oh!

codelipenghui Oct 31, 2022 Collaborator

Uh oh!

startjava Oct 31, 2022 Author

Uh oh!

momo-jun Nov 1, 2022 Collaborator

startjava
Oct 31, 2022

Replies: 4 comments 6 replies

codelipenghui
Oct 31, 2022
Collaborator

startjava Oct 31, 2022
Author

startjava Oct 31, 2022
Author

startjava
Oct 31, 2022
Author

startjava Oct 31, 2022
Author

codelipenghui
Oct 31, 2022
Collaborator

startjava
Oct 31, 2022
Author

codelipenghui Oct 31, 2022
Collaborator

startjava Oct 31, 2022
Author

momo-jun Nov 1, 2022
Collaborator