xuesongxs opened a new issue #14937:
URL: https://github.com/apache/pulsar/issues/14937
**Describe the bug**
Pulsar v2.8.1/v2.8.2
reconsumeLater can't receive message while restart all brokers.
Cluster:
1 broker
3 bookies
**To Reproduce**
Steps to reproduce the behavior:
1. create consumer:
```
public static void main(String[] args) throws Exception{
PulsarClient client = PulsarClient
.builder()
.serviceUrl("pulsar://127.0.0.1:6650")
.build();
ConsumerBuilder<NlMessage> consumerBuilder =
client.newConsumer(Schema.JSON(NlMessage.class)).topic("batchtest")
.subscriptionName("batchTest")
.subscriptionType(SubscriptionType.Shared)
.enableRetry(true)
.enableBatchIndexAcknowledgment(true)
.receiverQueueSize(1000)
.negativeAckRedeliveryDelay(20, TimeUnit.SECONDS)
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
// DLQ
DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder()
.deadLetterTopic("persistent://public/default/batchtest-dlq")
.retryLetterTopic("persistent://public/default/batchtest-retry")
.maxRedeliverCount(3)
.build();
consumerBuilder.deadLetterPolicy(deadLetterPolicy);
consumerBuilder.negativeAckRedeliveryDelay(10,
TimeUnit.MICROSECONDS);
// 需要在死信策略设置以后 再设置ackTimeout (如果ackTimeout == 0)
consumerBuilder.ackTimeout(0, TimeUnit.SECONDS); //
如果receiverQueueSize或者deadLetterPolicy后，ackTimeout又变为30s。
Consumer<NlMessage> consumer = consumerBuilder.subscribe();
while (true) {
Message<NlMessage> message = consumer.receive();
System.out.println("message:" + message.getMessageId());
consumer.reconsumeLater(message, 1, TimeUnit.SECONDS);
}
}
```
2. create producer:
```
public static void main(String[] args) throws PulsarClientException {
PulsarClient pulsarClient =
PulsarClient.builder().serviceUrl("pulsar://127.0.0.1:6650").build();
Producer<NlMessage> producer =
pulsarClient.newProducer(Schema.JSON(NlMessage.class))
.topic("persistent://public/default/batchtest")
.blockIfQueueFull(true)
.sendTimeout(0, TimeUnit.SECONDS)
.create();
long i = 1;
while(i < 1000000L){
producer.sendAsync(new NlMessage());
i++;
}
}
```
3. start consumer
4. start producer
5. stop broker
6. start broker
7. See error
**Expected behavior**
Continue receiving messages.
**Screenshots**
If applicable, add screenshots to help explain your problem.
**Desktop (please complete the following information):**
- OS: [e.g. iOS]
**Additional context**
Add any other context about the problem here.
--
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]
