You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by bo...@apache.org on 2022/04/08 08:01:17 UTC
[pulsar] branch master updated: [fix][broker] Fix rewind failed when ``redeliverUnacknowledgedMessages`` (#15046)
This is an automated email from the ASF dual-hosted git repository.
bogong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 22d6beb15b9 [fix][broker] Fix rewind failed when ``redeliverUnacknowledgedMessages`` (#15046)
22d6beb15b9 is described below
commit 22d6beb15b9d44e630e46931f5905d19f33fe724
Author: Qiang Zhao <74...@users.noreply.github.com>
AuthorDate: Fri Apr 8 16:01:11 2022 +0800
[fix][broker] Fix rewind failed when ``redeliverUnacknowledgedMessages`` (#15046)
### Motivation
According to many PRs flaky-test ``SimpleProducerConsumerTest#testRedeliveryFailOverConsumer``, the broker logs as below.
```
2022-04-07T11:19:16,561+0800 [pulsar-io-308-1] INFO org.apache.pulsar.broker.service.PulsarCommandSenderImpl - Send message to consumer, message id 3/0 epoch 0
2022-04-07T11:19:16,562+0800 [pulsar-client-io-336-1] INFO org.apache.pulsar.client.impl.ConsumerImpl - receiver queue received message, message id 3:0:-1 epoch 0
2022-04-07T11:19:16,562+0800 [pulsar-io-308-1] INFO org.apache.pulsar.broker.service.PulsarCommandSenderImpl - Send message to consumer, message id 3/1 epoch 0
2022-04-07T11:19:16,562+0800 [pulsar-client-io-336-1] INFO org.apache.pulsar.client.impl.ConsumerImpl - receiver queue received message, message id 3:1:-1 epoch 0
2022-04-07T11:19:16,576+0800 [pulsar-io-308-1] INFO org.apache.pulsar.broker.service.PulsarCommandSenderImpl - Send message to consumer, message id 3/2 epoch 0
2022-04-07T11:19:16,576+0800 [pulsar-io-308-1] INFO org.apache.pulsar.broker.service.PulsarCommandSenderImpl - Send message to consumer, message id 3/3 epoch 0
2022-04-07T11:19:16,576+0800 [pulsar-io-308-1] INFO org.apache.pulsar.broker.service.PulsarCommandSenderImpl - Send message to consumer, message id 3/4 epoch 0
2022-04-07T11:19:16,576+0800 [pulsar-io-308-1] INFO org.apache.pulsar.broker.service.PulsarCommandSenderImpl - Send message to consumer, message id 3/5 epoch 0
2022-04-07T11:19:16,576+0800 [pulsar-io-308-1] INFO org.apache.pulsar.broker.service.PulsarCommandSenderImpl - Send message to consumer, message id 3/6 epoch 0
2022-04-07T11:19:16,576+0800 [pulsar-io-308-1] INFO org.apache.pulsar.broker.service.PulsarCommandSenderImpl - Send message to consumer, message id 3/7 epoch 0
2022-04-07T11:19:16,576+0800 [pulsar-io-308-1] INFO org.apache.pulsar.broker.service.PulsarCommandSenderImpl - Send message to consumer, message id 3/8 epoch 0
2022-04-07T11:19:16,577+0800 [pulsar-client-io-336-1] INFO org.apache.pulsar.client.impl.ConsumerImpl - receiver queue received message, message id 3:2:-1 epoch 0
2022-04-07T11:19:16,577+0800 [pulsar-client-io-336-1] INFO org.apache.pulsar.client.impl.ConsumerImpl - receiver queue received message, message id 3:3:-1 epoch 0
2022-04-07T11:19:16,577+0800 [pulsar-client-io-336-1] INFO org.apache.pulsar.client.impl.ConsumerImpl - receiver queue received message, message id 3:4:-1 epoch 0
2022-04-07T11:19:16,577+0800 [main] INFO org.apache.pulsar.client.api.SimpleProducerConsumerTest - ----Consumer receive and ack message 3/0 epoch 0
2022-04-07T11:19:16,577+0800 [main] INFO org.apache.pulsar.client.api.SimpleProducerConsumerTest - ----Consumer receive and ack message 3/1 epoch 0
2022-04-07T11:19:16,577+0800 [main] INFO org.apache.pulsar.client.api.SimpleProducerConsumerTest - ----Consumer receive and ack message 3/2 epoch 0
2022-04-07T11:19:16,577+0800 [main] INFO org.apache.pulsar.client.api.SimpleProducerConsumerTest - ----Consumer receive and ack message 3/3 epoch 0
2022-04-07T11:19:16,577+0800 [main] INFO org.apache.pulsar.client.api.SimpleProducerConsumerTest - ----Trigger unack messages redeliver and clear receive queue----
2022-04-07T11:19:16,577+0800 [pulsar-client-io-336-1] INFO org.apache.pulsar.client.impl.ConsumerImpl - receiver queue received message, message id 3:5:-1 epoch 0
2022-04-07T11:19:16,577+0800 [pulsar-client-io-336-1] INFO org.apache.pulsar.client.impl.ConsumerImpl - receiver queue received message, message id 3:6:-1 epoch 0
2022-04-07T11:19:16,577+0800 [pulsar-client-io-336-1] INFO org.apache.pulsar.client.impl.ConsumerImpl - receiver queue received message, message id 3:7:-1 epoch 0
2022-04-07T11:19:16,577+0800 [pulsar-client-io-336-1] INFO org.apache.pulsar.client.impl.ConsumerImpl - receiver queue received message, message id 3:8:-1 epoch 0
2022-04-07T11:19:16,578+0800 [broker-topic-workers-OrderedScheduler-8-0] INFO org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumer - [persistent://my-property/my-ns/unacked-topic / subscriber-1-Consumer{subscription=PersistentSubscription{topic=persistent://my-property/my-ns/unacked-topic, name=subscriber-1}, consumerId=0, consumerName=dde4a, address=/127.0.0.1:49153}] Ignoring reDeliverUnAcknowledgedMessages: cancelPendingRequest on cursor failed
2022-04-07T11:19:16,578+0800 [pulsar-io-308-1] INFO org.apache.pulsar.broker.service.PulsarCommandSenderImpl - Send message to consumer, message id 3/9 epoch 0
2022-04-07T11:19:16,578+0800 [pulsar-client-io-336-1] INFO org.apache.pulsar.client.impl.ConsumerImpl - receiver queue received message, message id 3:9:-1 epoch 0
2022-04-07T11:19:17,580+0800 [main] WARN org.apache.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://my-property/my-ns/unacked-topic], messageId : [3:5:-1:0], messageConsumerEpoch : [0], consumerEpoch : [1]
2022-04-07T11:19:17,580+0800 [main] WARN org.apache.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://my-property/my-ns/unacked-topic], messageId : [3:6:-1:0], messageConsumerEpoch : [0], consumerEpoch : [1]
2022-04-07T11:19:17,580+0800 [main] WARN org.apache.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://my-property/my-ns/unacked-topic], messageId : [3:7:-1:0], messageConsumerEpoch : [0], consumerEpoch : [1]
2022-04-07T11:19:17,580+0800 [main] WARN org.apache.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://my-property/my-ns/unacked-topic], messageId : [3:8:-1:0], messageConsumerEpoch : [0], consumerEpoch : [1]
2022-04-07T11:19:17,580+0800 [main] WARN org.apache.pulsar.client.impl.ConsumerBase - Consumer filter old epoch message, topic : [persistent://my-property/my-ns/unacked-topic], messageId : [3:9:-1:0], messageConsumerEpoch : [0], consumerEpoch : [1]
```
In normal logic, the consumer will continue to receive messages from the broker. When the consumer triggers `redeliverUnacknowledgedMessages`, it will immediately increase the consumer epoch, and the request will rewind the broker cursor to trigger the broker to redeliver these messages with the new epoch.
And then, the consumer will filter old messages by epoch.
But in this case, we will find the following abnormal log, which means that the cursor pending read has been cancelled(cursor.cancelPendingReadRequest() return false), and then havePendingRead is not set to false. causing no cursor rewind at the end and didn't trigger the Dispatcher to re-push the messages.
> Ignoring reDeliverUnAcknowledgedMessages: cancelPendingRequest on cursor failed
Due to this abnormal behaviour causing the consumer will never receive redelivered messages with epoch 1 as below.
```
message id 3:4:-1 epoch 1
message id 3:5:-1 epoch 1
message id 3:6:-1 epoch 1
message id 3:7:-1 epoch 1
message id 3:8:-1 epoch 1
message id 3:9:-1 epoch 1
```
Relative code:
https://github.com/apache/pulsar/blob/81da8d3cd199fd6c1e4510a1c1c2ac71418efd5e/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java#L317-L328
https://github.com/apache/pulsar/blob/81da8d3cd199fd6c1e4510a1c1c2ac71418efd5e/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java#L136-L141
https://github.com/apache/pulsar/blob/81da8d3cd199fd6c1e4510a1c1c2ac71418efd5e/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java#L847-L851
### Modifications
- Force to rewind cursor when `redeliverUnacknowledgedMessages`
---
.../PersistentDispatcherSingleActiveConsumer.java | 18 ++++++------------
.../pulsar/client/api/SimpleProducerConsumerTest.java | 6 +++---
2 files changed, 9 insertions(+), 15 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
index e2f2ac9a9f4..c0effad9bbd 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
@@ -313,19 +313,13 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
name, consumer);
return;
}
-
- cancelPendingRead();
-
- if (!havePendingRead) {
- cursor.rewind();
- if (log.isDebugEnabled()) {
- log.debug("[{}-{}] Cursor rewinded, redelivering unacknowledged messages. ", name, consumer);
- }
- readMoreEntries(consumer);
- } else {
- log.info("[{}-{}] Ignoring reDeliverUnAcknowledgedMessages: cancelPendingRequest on cursor failed", name,
- consumer);
+ cursor.cancelPendingReadRequest();
+ havePendingRead = false;
+ cursor.rewind();
+ if (log.isDebugEnabled()) {
+ log.debug("[{}-{}] Cursor rewinded, redelivering unacknowledged messages. ", name, consumer);
}
+ readMoreEntries(consumer);
}
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
index 7506a9bf3f0..fa1505f4d6f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
@@ -2435,7 +2435,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
Message<byte[]> msg;
List<Message<byte[]>> messages1 = Lists.newArrayList();
for (int i = 0; i < consumeMsgInParts; i++) {
- msg = consumer.receive(RECEIVE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ msg = consumer.receive();
if (msg != null) {
messages1.add(msg);
consumer.acknowledge(msg);
@@ -2451,7 +2451,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
// (1.b) consume second consumeMsgInParts msgs and trigger redeliver
messages1.clear();
for (int i = 0; i < consumeMsgInParts; i++) {
- msg = consumer.receive(RECEIVE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ msg = consumer.receive();
if (msg != null) {
messages1.add(msg);
consumer.acknowledge(msg);
@@ -2474,7 +2474,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
int remainingMsgs = (2 * receiverQueueSize) - (2 * consumeMsgInParts);
messages1.clear();
for (int i = 0; i < remainingMsgs; i++) {
- msg = consumer.receive(RECEIVE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ msg = consumer.receive();
if (msg != null) {
messages1.add(msg);
consumer.acknowledge(msg);