[fix] [broker]Consumption stuck due to orphan consumers after the connection reconnect

[poorbarcode]

Motivation

There may leave an orphan consumer after the ServerCnx closed, which would cause consumption to get stuck. For example:

time	add a new consumer	connection health check
1		Health check is failed
2		Push the task ServerCnx.close into the eventLoop
3	Push a task typed cmd subscribe into the eventLoop
4		connection inactive
5		Close all consumers bound to this connection
6	Register new consumers into the subscription
7	send messages to this consumer(the send will fail, but no warn log[1])
8	These messages can not be delivered until the topic is unloaded

[1]: see [improve] [broker] Improve dispatcher log to trace io error that may cause consumption stuck #20568

Modifications

Just like producer registration^[2]: add a checker connection.isAtive after the consumer created complete.

[2]: https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java#LL1601C45-L1601C45

Documentation

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

Matching PR in forked repository

PR in forked repository: x

[codelipenghui]

Nice catch!

[merlimat]

👍 Nice catch and the fix looks good.

I wonder if we should force-close the connection and clear the consumer as well, when there is the socket write failure because the connection was already closed.

[poorbarcode]

Close this PR because the logic which this PR does already exists: https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L894-L915

[poorbarcode]

I wonder if we should force-close the connection and clear the consumer as well, when there is the socket write failure because the connection was already closed.

I did it in the PR #20568: do a health check immediately(if the health check fail, the connection will be closed) if got an IO error when sending messages to the client.