[fix][broker] Fix one potential to add duplicated consumer

[TakaHiR07]

Fixes #20576

Motivation

There is a case resulted in duplicated consumer, if we add the same consumer twice continuously, then close consumer, the offline consumer would remain in consumerList, because AbstractDispatcherMultipleConsumers#consumerList is arrayList and AbstractDispatcherMultipleConsumers#consumerSet is set.

There is a concurrent problem in ServerCnx#handleSubscribe, which cause this case. Suppose client request handleSubscribe() three times and request handleCloseConsumer(). The execute order is as follow. since the order is addConsumer->addConsumer->close, but not addConsumer->close->addConsumer, duplicated consumer problem occur.

Modifications

1. remove the step 9 "consumersLongHashmap.remove consumerFuture3".

• Because if judge existingConsumerFuture is not null, it means there is an earlier request try to handleSubscribe. the remove operation should be executed in the earlier request.

2. add a test to verify this concurrent problem and the fix

Verifying this change

• Make sure that the change passes the CI checks.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: TakaHiR07#10

[eolivelli]

You say in the linked issue and you have seen this twice.
Are you able to reproduce it manually and verify that with this fix the problem is really fixed?

[TakaHiR07]

You say in the linked issue and you have seen this twice. Are you able to reproduce it manually and verify that with this fix the problem is really fixed?

I guess this is one of the possibilities for duplicated consumer. And this pr is corresponding to our production case. I can't affirm the duplicated consumer problem do not occur anymore.

[TakaHiR07]

@poorbarcode @eolivelli @lhotari @congbobo184 Can you help review this pr? It is another possibilities besides #15051

[poorbarcode]

There are three clients in your example, the pulsar client maintains its own connection pool, which means the client1 will not use the same connection as client2 or client3.

The code you changed is the code of ServerCnx(this object is not shared across multi clients), so it will not affect other consumers of other clients

[poorbarcode]

Additional background:

If you have two pulsar clients, the consumers registered will be like this:

• pulsar-client-1
  • connection-1 (one-to-one relationship with ServerCnx)
    • consumer-1
    • consumer-2
  • connection-2
    • consumer-3
    • consumer-4
• pulsar-client-2
  • connection-3
    • consumer-5
    • consumer-6
  • connection-4
    • consumer-7
    • consumer-8

[poorbarcode]

So only consumer-1 and consumer-2 can be in conflict since their consumer-id is not the same, so all things are OK.

[TakaHiR07]

I guess it is one client request three times. Because in our server log, "Subscribing on topic" occur three times in a short time from the same remoteAddress host:ip. Therefore they are in the same connection ?

[poorbarcode]

I guess it is one client request three times. Because in our server log, "Subscribing on topic" occur three times in a short time from the same remoteAddress host:ip. Therefore they are in the same connection ?

could you provide the logs?

[poorbarcode]

By the log, we can not confirm if these three consumers are the same, because we can not know their consumer id. This log was been improved in the PR https://github.com/apache/pulsar/pull/20568/files#diff-1e0e8195fb5ec5a6d79acbc7d859c025a9b711f94e6ab37c94439e99b3202e84R1168

[poorbarcode]

time	req-1	req-2	close connection
1		put a new consumerFuture
2			close connection
3			mark the consumerFuture as failed
4			close the connection
5	got consumerFuture(failed)
6	remove the failed future
7		add the consumer into the list and set(list.size = 1 and set.size = 1)
8	put the second consumerFuture
9	add the second consumer into the list and set(list.size = 2 and set.size = 1)
10		remove the consumer from the list and set(list.size = 1 and set.size = 0)

Do you want to say the three requests executed as above?

• before the step-4 and after step-4, there are two different connections
• if the req-1 runs at the first connection, it will trigger another consumer.close. see: https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L896
  • (Highlight)But maybe left an orphan consumer in the consumer list, see: https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java#L203. I think you can write a test to reproduce this case.
• if the req-1 runs at the second connection, all things are OK. Just like the comment above [fix][broker] Fix one potential to add duplicated consumer #20583 (comment)

[TakaHiR07]

Yes

[poorbarcode]

BTW, there is a PR trying to fix the concurrent call subscribe in the same client

• [fix] [client] Messages lost when consumer reconnect #20591

I still think the current PR is trying to solve this issue is meaningful, and it would be nice to have a test that can reproduce it.

[poorbarcode]

@TakaHiR07

Could you take a look at this PR #22283 , #22270 and modify the current one? If yes, I will close #22283, #22270 ❤️

[TakaHiR07]

Add a test to verify this concurrent case and the fix. PTAL, thanks. @codelipenghui @eolivelli @lhotari @poorbarcode

[github-actions]

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

[graysonzeng]

I had the same problem on flink task and got fixed after trying this PR.

[TakaHiR07]

@poorbarcode @codelipenghui It seems this problem is not fixed in the master branch.

[TakaHiR07]

#22283 has fixed the same problem. Close this pr.