[fix][broker] The offline consumer is not removed from the consumerList #14970

lordcheng10 · 2022-03-31T09:55:28Z

Motivation

We found a problem: when the consumer was removed, it was not removed from the consumerList, but it was removed from the consumerSet and consumers：

The log to remove the consumer is as follows:

10:31:40.404 [BookKeeperClientWorker-OrderedExecutor-30-0] INFO org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers - Removed consumer Consumer{subscription=PersistentSubscription{topic=persistent://teg_onion_onion_logcenter_data_gz/teg_onion_onion_logcenter_data_gz/teg_onion_onion_logcenter_data_gz-partition-51, name=t_teg_onion_b_teg_onion_onion_logcenter_data_gz_cg_consumer_001}, consumerId=51, consumerName=ff7f4103a3, address=/11.181.202.252:55902} with pending 0 acks

Documentation

Check the box below or label this PR directly (if you have committer privilege).

Need to update docs?

doc-required

(If you need help on updating docs, create a doc issue)
no-need-doc

(Please explain why)
doc

(If this PR contains doc changes)

xuesongxs · 2022-03-31T11:25:51Z

#13787
Just like this bug.

lhotari

Good catch. I wonder if it would be better to prevent adding duplicates to the list and doing a check in addConsumer? I think that would be a better solution to the problem.
I also have a suggestion to use removeIf there isn't a way to prevent duplicates getting added to the list.

lhotari · 2022-03-31T16:48:47Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

    @Override
    public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceException {
        // decrement unack-message count for removed consumer
        addUnAckedMessages(-consumer.getUnackedMessages());
        if (consumerSet.removeAll(consumer) == 1) {
-            consumerList.remove(consumer);
+            removeAllConsumer(consumer);


The removeAllConsumer doesn't look great. I wonder if using .removeIf would be better:

Suggested change

removeAllConsumer(consumer);

consumerList.removeIf(consumer::equals);

I agree with this sentiment. I wonder about the data structure itself, too. Why is it a list and not a set?

+1 to @lhotari 's suggestion

I agree to @lhotari 's suggestion

Fixed:
1.do a check in addConsumer;
2.use removeIf;

PTAL,thanks! @lhotari @eolivelli

@lordcheng10 there's a minor checkstyle issue in the recent changes. please check that

Fixed.
PTAL,thanks! @lhotari

lordcheng10 · 2022-03-31T18:10:34Z

In fact, the problem here may be the problem of consumer equality, similar to the previous producer problem.

When there are duplicate consumers in the consumerList, it is possible to remove the actually successfully registered consumers when removing them.

In our scenario, we also found a problem: when the consumer reconnects, the broker no longer pushes messages to these consumers. Through the stats command, we see these The consumer's permit is 0, and it is a consumer that has been removed from the consumerSet and consumers. The wrong consumer could be removed .

So the fundamental problem here is actually the definition of consumer equality. I think we might need to do something like #12846 on the definition of equality for consumers.

lordcheng10 · 2022-03-31T18:11:07Z

I think we also need to implement a method similar to isSuccessorTo to determine whether the consumer can be overridden, what do you think? @lhotari @michaeljmarshall @eolivelli

like this:

public boolean isSuccessorTo(Consumer other){
    return Objects.equals(cnx.clientAddress(), other.cnx.clientAddress()) && consumerId == other.consumerId
            &&  other.consumerEpoch < consumerEpoch;
}

michaeljmarshall · 2022-03-31T18:50:54Z

@lordcheng10 - that is a great finding. I think it seems very reasonable to adjust the consumer logic, but I still haven't researched the consumer remove logic, so I am not authoritative on the subject. I wonder if we'd be able to add a test that shows the isSuccessorTo logic is necessary.

2.use removeIf;

eolivelli · 2022-04-01T06:44:20Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

+            log.warn("[{}] Consumer with the same id is already created:"
+                            + " consumerId={}, consumer={}",
+                    consumer.cnx().clientAddress(), consumer.consumerId(), consumer);
+            throw new BrokerServiceException("Consumer with the same id is already created!");


what about using ConsumerBusyException ?

eolivelli · 2022-04-01T06:44:30Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

+                            + " consumerId={}, consumer={}",
+                    consumer.cnx().clientAddress(), consumer.consumerId(), consumer);
+            throw new BrokerServiceException("Consumer with the same id is already created!");
+


nit: remove useless newline

Fixed. PTAL,thanks! @eolivelli

2.remove useless newline;

eolivelli

LGTM

although adding a test will help in understanding how to reproduce the problem

lordcheng10 · 2022-04-01T08:06:40Z

/pulsarbot run-failure-checks

lordcheng10 · 2022-04-01T08:17:13Z

/pulsarbot run-failure-checks

Jason918 · 2022-04-01T09:32:49Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

@@ -153,6 +153,13 @@ public synchronized void addConsumer(Consumer consumer) throws BrokerServiceExce
            throw new ConsumerBusyException("Subscription reached max consumers limit");
        }

+        if (consumerList.contains(consumer)) {


Can you provide the case of duplicated consumer?
I wonder if we should replace the old consumer with the new one. The older consumer maybe somewhat expired.

Yes so I think we should write a method like org.apache.pulsar.broker.service.Producer#isSuccessorTo to judge whether the previous consumer object should be overwritten by consumerEpoch，like this:

public boolean isSuccessorTo(Consumer other){
return Objects.equals(cnx.clientAddress(), other.cnx.clientAddress()) && consumerId == other.consumerId
&& other.consumerEpoch < consumerEpoch;
}

Can you provide the case of duplicated consumer?

OK, I will try to find duplicate cases.

lordcheng10 · 2022-04-01T10:59:52Z

/pulsarbot run-failure-checks

github-actions · 2022-05-02T02:16:11Z

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

github-actions · 2022-06-02T02:13:33Z

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

remove all consumer from consumerList

4d3a3ae

github-actions bot assigned lordcheng10 Mar 31, 2022

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

Technoboy- changed the title ~~Fix:The offline consumer is not removed from the consumerList~~ [fix][broker] The offline consumer is not removed from the consumerList Mar 31, 2022

lhotari requested changes Mar 31, 2022

View reviewed changes

lhotari requested review from michaeljmarshall and eolivelli March 31, 2022 16:53

lordcheng10 added 2 commits April 1, 2022 10:46

1.do a check in addConsumer;

eb75a3c

2.use removeIf;

check style

fc1d0b5

eolivelli reviewed Apr 1, 2022

View reviewed changes

1.use ConsumerBusyException;

e1c9596

2.remove useless newline;

eolivelli approved these changes Apr 1, 2022

View reviewed changes

lordcheng10 requested a review from lhotari April 1, 2022 08:06

Jason918 reviewed Apr 1, 2022

View reviewed changes

This was referenced Apr 6, 2022

Broker registered the same consumers #15050

Closed

[fix][broker] Fix potential to add duplicated consumer #15051

Merged

sijie mentioned this pull request Apr 6, 2022

ISSUE-15050: Broker registered the same consumers streamnative/pulsar-archived#4029

Open

github-actions bot added the lifecycle/stale label May 2, 2022

lhotari mentioned this pull request Jul 15, 2022

[Java Client] Send CloseConsumer on client timeout #16616

Merged

1 task

TakaHiR07 mentioned this pull request Jun 14, 2023

[Bug] [broker] subscription of topic in a permanent fence state and consumerList remain offline consumer #20576

Closed

2 tasks

Technoboy- added this to the 3.2.0 milestone Jul 31, 2023

Technoboy- modified the milestones: 3.2.0, 3.3.0 Dec 22, 2023

coderzc modified the milestones: 3.3.0, 3.4.0 May 8, 2024

lhotari modified the milestones: 4.0.0, 4.1.0 Oct 14, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[fix][broker] The offline consumer is not removed from the consumerList #14970

[fix][broker] The offline consumer is not removed from the consumerList #14970

lordcheng10 commented Mar 31, 2022 •

edited

Loading

xuesongxs commented Mar 31, 2022

lhotari left a comment

lhotari Mar 31, 2022

michaeljmarshall Mar 31, 2022

eolivelli Mar 31, 2022

lordcheng10 Mar 31, 2022 •

edited

Loading

lordcheng10 Apr 1, 2022 •

edited

Loading

lhotari Apr 1, 2022

lordcheng10 Apr 1, 2022

lordcheng10 commented Mar 31, 2022 •

edited

Loading

lordcheng10 commented Mar 31, 2022 •

edited

Loading

michaeljmarshall commented Mar 31, 2022

eolivelli Apr 1, 2022

lordcheng10 Apr 1, 2022

eolivelli Apr 1, 2022

lordcheng10 Apr 1, 2022

eolivelli left a comment

lordcheng10 commented Apr 1, 2022

lordcheng10 commented Apr 1, 2022

Jason918 Apr 1, 2022

lordcheng10 Apr 1, 2022 •

edited

Loading

lordcheng10 Apr 1, 2022

lordcheng10 commented Apr 1, 2022

github-actions bot commented May 2, 2022

github-actions bot commented Jun 2, 2022

	removeAllConsumer(consumer);
	consumerList.removeIf(consumer::equals);

[fix][broker] The offline consumer is not removed from the consumerList #14970

Are you sure you want to change the base?

[fix][broker] The offline consumer is not removed from the consumerList #14970

Conversation

lordcheng10 commented Mar 31, 2022 • edited Loading

Motivation

Documentation

xuesongxs commented Mar 31, 2022

lhotari left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lordcheng10 Mar 31, 2022 • edited Loading

Choose a reason for hiding this comment

lordcheng10 Apr 1, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lordcheng10 commented Mar 31, 2022 • edited Loading

lordcheng10 commented Mar 31, 2022 • edited Loading

michaeljmarshall commented Mar 31, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eolivelli left a comment

Choose a reason for hiding this comment

lordcheng10 commented Apr 1, 2022

lordcheng10 commented Apr 1, 2022

Choose a reason for hiding this comment

lordcheng10 Apr 1, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lordcheng10 commented Apr 1, 2022

github-actions bot commented May 2, 2022

github-actions bot commented Jun 2, 2022

lordcheng10 commented Mar 31, 2022 •

edited

Loading

lordcheng10 Mar 31, 2022 •

edited

Loading

lordcheng10 Apr 1, 2022 •

edited

Loading

lordcheng10 commented Mar 31, 2022 •

edited

Loading

lordcheng10 commented Mar 31, 2022 •

edited

Loading

lordcheng10 Apr 1, 2022 •

edited

Loading