[fix][client] Fix duplicate messages caused by seek #16171

nodece · 2022-06-22T09:15:06Z

Signed-off-by: Zixuan Liu nodeces@gmail.com

Motivation

When subscribing to a message on a partitioned topic, do the seek operation and then consume the messages, which sometimes will receive duplicate messages.

The root cause is that when a seek operation is performed if have the task in the pendingReceives queue, we will get old messages from before doing the seek operation.

Modifications

Add a seek check to the logic of the received message, when the seek operation is in progress, skip put the message to incomingMessages queue

Verifying this change

org.apache.pulsar.broker.service.SubscriptionSeekTest#testSeekByFunctionAndMultiTopic cover this changes.

Documentation

doc-not-needed

nodece · 2022-06-23T02:22:22Z

/pulsarbot rerun-failure-checks

BewareMyPower · 2022-06-23T08:31:53Z

org.apache.pulsar.broker.service.SubscriptionSeekTest#testSeekByFunctionAndMultiTopic cover this changes.

Why didn't it fail before?

nodece · 2022-06-23T08:56:55Z

org.apache.pulsar.broker.service.SubscriptionSeekTest#testSeekByFunctionAndMultiTopic cover this changes.

Why didn't it fail before?

@BewareMyPower I don't know how to explain, the master branch cannot reproduce this issue, but the #15568 can reproduce this issue.

BewareMyPower · 2022-06-23T09:00:18Z

Did you mean after applying the changes of #15568 will this test fail?

nodece · 2022-06-23T10:26:30Z

Did you mean after applying the changes of #15568 will this test fail?

Yes, could you take a look this issue?

BewareMyPower · 2022-06-23T10:37:20Z

Sure.

pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java

BewareMyPower · 2022-06-28T07:38:34Z

Yes this PR fixes the race condition when seek is called after subscribe, but I think there is another bug that subscribe doesn't looks like a synchronous operation. Without this PR, the test could still pass after sleeping for a while before seek.

        org.apache.pulsar.client.api.Consumer<String> consumer = pulsarClient
                .newConsumer(Schema.STRING).startMessageIdInclusive()
                .topics(Arrays.asList(topicName, topicName2)).subscriptionName("my-sub").subscribe();

        Thread.sleep(2000);

        consumer.seek((partitionedTopic) -> {

I think the root cause is that MultiTopicsConsumerImpl#subscribeFuture() completed too early.

nodece · 2022-06-28T09:28:30Z

subscribe is an async task, when you sleep 2s, it means we have received all messages from the broker. If you remove this sleep, there are some messages coming to the client from the broker.

Signed-off-by: Zixuan Liu <nodeces@gmail.com>

…r#16171

…ting logic for better handle the checkpoint. (#19972) * Change the initial start cursor and stop cursor to better handle the consuming behaviors. * Create the initial subscription instead seek every time. This should fix the wrong position setting. * Fix the wrong stop cursor, make sure it stops at the correct space * Drop Consumer.seek() for apache/pulsar#16171

…ting logic for better handle the checkpoint. (apache#19972) * Change the initial start cursor and stop cursor to better handle the consuming behaviors. * Create the initial subscription instead seek every time. This should fix the wrong position setting. * Fix the wrong stop cursor, make sure it stops at the correct space * Drop Consumer.seek() for apache/pulsar#16171 (cherry picked from commit 18d21a0)

…ting logic for better handle the checkpoint. (apache#19972) * Change the initial start cursor and stop cursor to better handle the consuming behaviors. * Create the initial subscription instead seek every time. This should fix the wrong position setting. * Fix the wrong stop cursor, make sure it stops at the correct space * Drop Consumer.seek() for apache/pulsar#16171

…ting logic for better handle the checkpoint. (#19972) (#20565) * Change the initial start cursor and stop cursor to better handle the consuming behaviors. * Create the initial subscription instead seek every time. This should fix the wrong position setting. * Fix the wrong stop cursor, make sure it stops at the correct space * Drop Consumer.seek() for apache/pulsar#16171

…ting logic for better handle the checkpoint. (#19972) (#20564) * Change the initial start cursor and stop cursor to better handle the consuming behaviors. * Create the initial subscription instead seek every time. This should fix the wrong position setting. * Fix the wrong stop cursor, make sure it stops at the correct space * Drop Consumer.seek() for apache/pulsar#16171

github-actions · 2022-08-25T02:19:07Z

The pr had no activity for 30 days, mark with Stale label.

poorbarcode · 2022-10-11T06:23:18Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java

@@ -252,6 +252,10 @@ private void receiveMessageFromConsumer(ConsumerImpl<T> consumer, boolean batchR
            messagesFuture = consumer.receiveAsync().thenApply(Collections::singletonList);
        }
        messagesFuture.thenAcceptAsync(messages -> {
+            if (consumer.isDuringSeek()) {
+                receiveMessageFromConsumer(consumer, batchReceive);


When incomingMessages is not empty, there are two possibilities:

loop until there is no message in the incomingMessages

If ConsumerImpl.seek fails, the messages have popped from incomingMessages cannot be consumed until redeliver executes

I also have a question about this part.

It looks like not only from the MultiTopicConsumer. If it is from the user side, using a consumer to receive the messages and another thread try to seek the subscription to another position, they will also receive the duplicated messages right?

2. If ConsumerImpl.seek fails, the messages have popped from incomingMessages cannot be consumed until redeliver executes

Good catch! We need to consider this.

It looks like not only from the MultiTopicConsumer.

Yes.

If it is from the user side, using a consumer to receive the messages and another thread try to seek the subscription to another position, they will also receive the duplicated messages right?

For MultiTopicConsumer, this is right, because the MultiTopicConsumer has a loop to pulling the messsage.

Ok, I guess the fix can only fix the case that the MultiTopicConsumer poll messages from the internal consumer, but it can't fix the issue that the user facing. After the message polled from the internal consumer to the queue of the MultiTopicConsumer, the user will still have chance to get duplicated messages during the seek operation?

Ok, I guess the fix can only fix the case that the MultiTopicConsumer poll messages from the internal consumer.

Right.

the user will still have chance to get duplicated messages during the seek operation?

I think we need to figure out the details of this seek.

When to clean incomingMessages, after or before seek?

During the seek, can the client continue to consume the incomingMessages?

…ting logic for better handle the checkpoint. (apache#19972) * Change the initial start cursor and stop cursor to better handle the consuming behaviors. * Create the initial subscription instead seek every time. This should fix the wrong position setting. * Fix the wrong stop cursor, make sure it stops at the correct space * Drop Consumer.seek() for apache/pulsar#16171

…ting logic for better handle the checkpoint. (#19972) * Change the initial start cursor and stop cursor to better handle the consuming behaviors. * Create the initial subscription instead seek every time. This should fix the wrong position setting. * Fix the wrong stop cursor, make sure it stops at the correct space * Drop Consumer.seek() for apache/pulsar#16171

nodece · 2023-03-25T17:45:44Z

No reviewer, so close this PR.

hicolour · 2023-05-02T13:56:29Z

Hey @nodece, why this PR was closed? Does it it mean that this PR and #15568 will not be delivered to the maistream ?

nodece · 2023-05-03T16:54:27Z

Hey @nodece, why this PR was closed?

Currently, there is no reviewer pushing this PR, so I decided to close this PR. If anyone is willing to push, I will reopen.

Does it it mean that this PR and #15568 will not be delivered to the maistream ?

Right!

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Jun 22, 2022

codelipenghui reviewed Jun 28, 2022

View reviewed changes

pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java Outdated Show resolved Hide resolved

codelipenghui assigned nodece Jun 28, 2022

codelipenghui added this to the 2.11.0 milestone Jun 28, 2022

BewareMyPower approved these changes Jun 28, 2022

View reviewed changes

nodece mentioned this pull request Jul 7, 2022

[PIP-150][improve][broker] Support read the message of startMessageId position on the broker side #15568

Closed

3 tasks

nodece requested review from codelipenghui, gaozhangmin, mattisonchao, hezhangjian, congbobo184 and gaoran10 July 7, 2022 03:32

nodece force-pushed the fix-duplicate-messages-seek branch from 79a9fae to 6b8ef55 Compare July 12, 2022 14:27

[fix][client] Fix duplicate messages caused by seek

25abae8

Signed-off-by: Zixuan Liu <nodeces@gmail.com>

nodece force-pushed the fix-duplicate-messages-seek branch from 6b8ef55 to 25abae8 Compare July 12, 2022 15:25

nodece requested review from RobertIndie and Technoboy- July 14, 2022 09:05

nodece mentioned this pull request Jul 25, 2022

PIP 194 : Pulsar client: seek command add epoch #16757

Open

codelipenghui modified the milestones: 2.11.0, 2.12.0 Jul 26, 2022

syhily added a commit to streamnative/flink that referenced this pull request Aug 11, 2022

[FLINK-27399][Connector/Pulsar] Drop Consumer.seek() for apache/pulsa…

2baa322

…r#16171

syhily added a commit to streamnative/flink that referenced this pull request Aug 12, 2022

[FLINK-27399][Connector/Pulsar] Drop Consumer.seek() for apache/pulsa…

f807c4a

…r#16171

syhily added a commit to streamnative/flink that referenced this pull request Aug 12, 2022

[FLINK-27399][Connector/Pulsar] Drop Consumer.seek() for apache/pulsa…

938be57

…r#16171

syhily added a commit to streamnative/flink that referenced this pull request Aug 12, 2022

[FLINK-27399][Connector/Pulsar] Drop Consumer.seek() for apache/pulsa…

7ea6b3c

…r#16171

github-actions bot added the Stale label Aug 25, 2022

poorbarcode reviewed Oct 11, 2022

View reviewed changes

nodece requested review from eolivelli and michaeljmarshall March 25, 2023 17:42

nodece closed this Mar 25, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[fix][client] Fix duplicate messages caused by seek #16171

[fix][client] Fix duplicate messages caused by seek #16171

nodece commented Jun 22, 2022 •

edited

Loading

nodece commented Jun 23, 2022

BewareMyPower commented Jun 23, 2022

nodece commented Jun 23, 2022

BewareMyPower commented Jun 23, 2022

nodece commented Jun 23, 2022

BewareMyPower commented Jun 23, 2022

BewareMyPower commented Jun 28, 2022

nodece commented Jun 28, 2022

github-actions bot commented Aug 25, 2022

poorbarcode Oct 11, 2022

codelipenghui Oct 13, 2022

nodece Oct 13, 2022

codelipenghui Oct 14, 2022

nodece Oct 14, 2022

nodece commented Mar 25, 2023

hicolour commented May 2, 2023

nodece commented May 3, 2023

[fix][client] Fix duplicate messages caused by seek #16171

[fix][client] Fix duplicate messages caused by seek #16171

Conversation

nodece commented Jun 22, 2022 • edited Loading

Motivation

Modifications

Verifying this change

Documentation

nodece commented Jun 23, 2022

BewareMyPower commented Jun 23, 2022

nodece commented Jun 23, 2022

BewareMyPower commented Jun 23, 2022

nodece commented Jun 23, 2022

BewareMyPower commented Jun 23, 2022

BewareMyPower commented Jun 28, 2022

nodece commented Jun 28, 2022

github-actions bot commented Aug 25, 2022

poorbarcode Oct 11, 2022

Choose a reason for hiding this comment

codelipenghui Oct 13, 2022

Choose a reason for hiding this comment

nodece Oct 13, 2022

Choose a reason for hiding this comment

codelipenghui Oct 14, 2022

Choose a reason for hiding this comment

nodece Oct 14, 2022

Choose a reason for hiding this comment

nodece commented Mar 25, 2023

hicolour commented May 2, 2023

nodece commented May 3, 2023

nodece commented Jun 22, 2022 •

edited

Loading