codelipenghui commented on a change in pull request #13707:
URL: https://github.com/apache/pulsar/pull/13707#discussion_r783152015
##########
File path:
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
##########
@@ -276,11 +276,10 @@ protected ConsumerImpl(PulsarClientImpl client, String
topic, ConsumerConfigurat
duringSeek = new AtomicBoolean(false);
if (conf.getAckTimeoutMillis() != 0) {
- if (conf.getTickDurationMillis() > 0) {
- this.unAckedMessageTracker = new UnAckedMessageTracker(client,
this, conf.getAckTimeoutMillis(),
- Math.min(conf.getTickDurationMillis(),
conf.getAckTimeoutMillis()));
+ if (conf.getAckTimeoutRedeliveryBackoff() != null) {
Review comment:
If users enabled this feature, we will use a different unack message
tracker to track the unack message, I think an easy and more efficient way to
implement the feature is
1. Introduce a `UnackMessageIdWrapper` that with Netty recycler, change the
`UnAckedMessageTracker.timePartitions` to
`ArrayDeque<ConcurrentOpenHashSet<UnackMessageIdWrapper>>`
2. The UnackMessageIdWrapper can have the redelivery count
3. Only add the timeout messageID to the redelivery tracker(the new
introduced map)
4. For each timer tick of the `UnAckedMessageTracker`, we should also check
if there messages need to redeliver from the redelivery tracker
--
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]
