[pulsar-broker] clean up active consumer on already closed connection #13196

rdhabalia · 2021-12-08T20:29:51Z

Motivation

We have been frequently seeing an issue with an exclusive subscription where the broker doesn't clean up consumers on closed on connection and because of that exclusive consumer keeps failing with ConsumerBusyException even though none of the consumers is actually connected to the broker.
Below log example shows that connection is closed but still broker couldn't clean up consumer in race condition when consumer quickly disconnects after creating the subscription and subsequent consumer creation requests are keep failing.

23:30:16.896 [pulsar-io-23-42] INFO  org.apache.pulsar.broker.service.ServerCnx - [/10.11.12.13:60851] Subscribing on topic persistent://my-prop/my-cluster/ns/topic / sub1
:
23:30:17.223 [pulsar-io-23-42] INFO  org.apache.pulsar.broker.service.ServerCnx - [/10.11.12.13:60851] Closed consumer, consumerId=20
:
23:30:17.291 [bookkeeper-ml-workers-OrderedExecutor-1-0] INFO  org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://my-prop/my-cluster/ns/topic][sub1] Created new subscription for 25
:
23:30:17.301 [pulsar-io-23-42] INFO  org.apache.pulsar.broker.service.ServerCnx - Closed connection from /10.11.12.13:60851
:
23:30:17.302 [pulsar-io-23-42] INFO  org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer - Removing consumer Consumer{subscription=PersistentSubscription{topic=persistent://my-prop/my-cluster/ns/topic, name=sub1}, consumerId=21, consumerName=c2302, address=/10.11.12.13:60851}
:
23:30:17.302 [bookkeeper-ml-workers-OrderedExecutor-1-0] INFO  org.apache.pulsar.broker.service.ServerCnx - [/10.11.12.13:60851] Created subscription on topic persistent://my-prop/my-cluster/ns/topic / sub1
:
:
:

23:30:17.496 [pulsar-io-23-36] WARN  org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://my-prop/my-cluster/ns/topic][sub1] Consumer 25 7a977 already connected
23:30:17.885 [pulsar-io-23-36] INFO  org.apache.pulsar.broker.service.ServerCnx - [/10.11.12.13:60853] Subscribing on topic persistent://my-prop/my-cluster/ns/topic / sub1
23:30:17.886 [pulsar-io-23-36] WARN  org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://my-prop/my-cluster/ns/topic][sub1] Consumer 25 7a977 already connected
23:30:18.637 [pulsar-io-23-36] INFO  org.apache.pulsar.broker.service.ServerCnx - [/10.11.12.13:60853] Subscribing on topic persistent://my-prop/my-cluster/ns/topic / sub1

Modification

Broker should clean up consumer if connection is already closed and allow consumer to reconnect as an active consumer.

github-actions · 2021-12-08T20:30:27Z

@rdhabalia:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

Jason918 · 2021-12-09T02:25:51Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

+                            consumer.close();
+                        }
+                    } catch (Exception be) {
+                        log.info("Failed to clean up consumer on closed connection {}, {}", consumer, be.getMessage());


Any reason this log is info level?

it's just info and doesn't have to be critical. therefore, log level is info.

I just want to make sure that if it's an exception goes here, the problem you mentioned won't occur,
as "subsequent consumer creation requests are keep failing" is totally worth to be an error.

To me. It seems like a bandaid. I think we should see how we landed in this situation. Perhaps we should have clean the consumer at the connection close itself. If there is a race condition we should the race condition itself.

@Jason918 changed to error log level.

@pkumar-singh this part handles resiliency to avoid consumer unavailability due to any kind of race condition.

Anonymitaet · 2021-12-09T02:27:07Z

@rdhabalia is this a bug fix? (do not need to update docs?)

Jason918 · 2021-12-10T02:20:01Z

@rdhabalia is this a bug fix? (do not need to update docs?)

Yes, a bug fix.

rdhabalia · 2021-12-20T19:23:27Z

ping. PR is ready to merge.
@Jason918 PTAL

rdhabalia · 2021-12-21T01:38:10Z

ping

Jason918 · 2021-12-21T01:53:23Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

+                        consumer = subscriptionFuture.isDone() ? getActiveConsumer(subscriptionFuture.get()) : null;
+                        // cleanup consumer if connection is already closed
+                        if (consumer != null && !consumer.cnx().isActive()) {
+                            consumer.close();


I my understanding, this consumer should be closed elsewhere in normal situation. This means there is a chance that consumer.close wii be called multi times. I think we should add closed state check in consumer.close to avoid other unexpected issues.

no need to check closed state because if consumer connection is not active and if consumer is still pretends as an active consumer then it requires cleanup by closing it immedaitely.

if consumer connection is not active and if consumer is still pretends as an active consumer then it requires cleanup by closing it immedaitely.

I get what you are trying to do, and I am +1 for this.

But do you agree that consumer.close may be called in two threads at the same time?
One is here, and the other may be in org.apache.pulsar.broker.service.ServerCnx#channelInactive.

it won't happen in this code path. however, in any case it would be safe to call close for already closed consumer,

however, in any case it would be safe to call close for already closed consumer,

Great to know. Thx

rdhabalia · 2021-12-21T22:34:44Z

ping.

Jason918

LGTM

rdhabalia · 2022-01-08T01:35:23Z

ping. can someone please review the PR?

eolivelli · 2022-01-08T21:41:18Z

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java

+                    latch.countDown();
+                }).subscribe();
+            } catch (Exception e) {
+                // Ok


Can we perform some checks on this Exception? At least match the Java class.
Otherwise here we can pass the tests even for other kinds of bugs

…#13196) ### Motivation We have been frequently seeing an issue with an exclusive subscription where the broker doesn't clean up consumers on closed on connection and because of that exclusive consumer keeps failing with `ConsumerBusyException` even though none of the consumers is actually connected to the broker. Below log example shows that connection is closed but still broker couldn't clean up consumer in race condition when consumer quickly disconnects after creating the subscription and subsequent consumer creation requests are keep failing. ``` 23:30:16.896 [pulsar-io-23-42] INFO org.apache.pulsar.broker.service.ServerCnx - [/10.11.12.13:60851] Subscribing on topic persistent://my-prop/my-cluster/ns/topic / sub1 : 23:30:17.223 [pulsar-io-23-42] INFO org.apache.pulsar.broker.service.ServerCnx - [/10.11.12.13:60851] Closed consumer, consumerId=20 : 23:30:17.291 [bookkeeper-ml-workers-OrderedExecutor-1-0] INFO org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://my-prop/my-cluster/ns/topic][sub1] Created new subscription for 25 : 23:30:17.301 [pulsar-io-23-42] INFO org.apache.pulsar.broker.service.ServerCnx - Closed connection from /10.11.12.13:60851 : 23:30:17.302 [pulsar-io-23-42] INFO org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer - Removing consumer Consumer{subscription=PersistentSubscription{topic=persistent://my-prop/my-cluster/ns/topic, name=sub1}, consumerId=21, consumerName=c2302, address=/10.11.12.13:60851} : 23:30:17.302 [bookkeeper-ml-workers-OrderedExecutor-1-0] INFO org.apache.pulsar.broker.service.ServerCnx - [/10.11.12.13:60851] Created subscription on topic persistent://my-prop/my-cluster/ns/topic / sub1 : : : 23:30:17.496 [pulsar-io-23-36] WARN org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://my-prop/my-cluster/ns/topic][sub1] Consumer 25 7a977 already connected 23:30:17.885 [pulsar-io-23-36] INFO org.apache.pulsar.broker.service.ServerCnx - [/10.11.12.13:60853] Subscribing on topic persistent://my-prop/my-cluster/ns/topic / sub1 23:30:17.886 [pulsar-io-23-36] WARN org.apache.pulsar.broker.service.persistent.PersistentTopic - [persistent://my-prop/my-cluster/ns/topic][sub1] Consumer 25 7a977 already connected 23:30:18.637 [pulsar-io-23-36] INFO org.apache.pulsar.broker.service.ServerCnx - [/10.11.12.13:60853] Subscribing on topic persistent://my-prop/my-cluster/ns/topic / sub1 ``` ### Modification Broker should clean up consumer if connection is already closed and allow consumer to reconnect as an active consumer. (cherry picked from commit 496afa7)

[pulsar-broker] clean up active consumer on already closed connection

9506226

rdhabalia added the area/broker label Dec 8, 2021

rdhabalia added this to the 2.10.0 milestone Dec 8, 2021

rdhabalia requested review from merlimat, sijie, jiazhai, eolivelli, codelipenghui and massakam December 8, 2021 20:29

rdhabalia self-assigned this Dec 8, 2021

github-actions bot added the doc-label-missing label Dec 8, 2021

Jason918 reviewed Dec 9, 2021

View reviewed changes

Anonymitaet added doc-not-needed Your PR changes do not impact docs and removed doc-label-missing labels Dec 14, 2021

fix logging level

c57ad2b

Jason918 reviewed Dec 21, 2021

View reviewed changes

Jason918 approved these changes Dec 22, 2021

View reviewed changes

eolivelli reviewed Jan 8, 2022

View reviewed changes

codelipenghui approved these changes Jan 21, 2022

View reviewed changes

codelipenghui added 3 commits January 25, 2022 11:43

Merge remote-tracking branch 'apache/master' into busy_cons

86e6c18

Apply comment

2a272a2

Fix typo

2647468

codelipenghui merged commit 496afa7 into apache:master Jan 25, 2022

codelipenghui added release/2.8.3 release/2.9.3 and removed release/2.8.3 labels Jan 25, 2022

codelipenghui added the cherry-picked/branch-2.9 Archived: 2.9 is end of life label Jan 29, 2022

gaoran10 added release/2.9.2 and removed release/2.9.3 labels Feb 16, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[pulsar-broker] clean up active consumer on already closed connection #13196

[pulsar-broker] clean up active consumer on already closed connection #13196

rdhabalia commented Dec 8, 2021

github-actions bot commented Dec 8, 2021

Jason918 Dec 9, 2021

rdhabalia Dec 9, 2021

Jason918 Dec 10, 2021

pkumar-singh Dec 14, 2021

rdhabalia Dec 16, 2021 •

edited

Loading

Anonymitaet commented Dec 9, 2021

Jason918 commented Dec 10, 2021

rdhabalia commented Dec 20, 2021

rdhabalia commented Dec 21, 2021

Jason918 Dec 21, 2021

rdhabalia Dec 21, 2021

Jason918 Dec 21, 2021

rdhabalia Dec 21, 2021

Jason918 Dec 22, 2021

rdhabalia commented Dec 21, 2021

Jason918 left a comment

rdhabalia commented Jan 8, 2022

eolivelli Jan 8, 2022

[pulsar-broker] clean up active consumer on already closed connection #13196

[pulsar-broker] clean up active consumer on already closed connection #13196

Conversation

rdhabalia commented Dec 8, 2021

Motivation

Modification

github-actions bot commented Dec 8, 2021

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

rdhabalia Dec 16, 2021 • edited Loading

Choose a reason for hiding this comment

Anonymitaet commented Dec 9, 2021

Jason918 commented Dec 10, 2021

rdhabalia commented Dec 20, 2021

rdhabalia commented Dec 21, 2021

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

Choose a reason for hiding this comment

rdhabalia commented Dec 21, 2021

Jason918 left a comment

Choose a reason for hiding this comment

rdhabalia commented Jan 8, 2022

Choose a reason for hiding this comment

rdhabalia Dec 16, 2021 •

edited

Loading