[fix][broker]A failed consumer/producer future in ServerCnx can never be removed

[poorbarcode]

Motivation

Conditions under which problems occur

• Failover/Exclusive subscription
• Sockets of Pulsar IO are not stable
• ServerCnx thread is busy.

time	registering consumer 1st	registering consumer 2nd
1	Registered
2	Socket is unstable
3	Closing ServerCnx
4		Retry to subscribe
5		Check whether the previous consumer is still active[1], if not, override it by the current registering
6	ServerCnx is closed
7		Call serverCnx.checkConnectionLiveness, the Completable Future returned will never be completed
8		Client-side: call ClientCnx.removeConsumer after the subscribing is timeout
9		There is a failed future in ServerCnx.consumers

[1]: AbstractDispatcherSingleActiveConsumer.java

 public synchronized CompletableFuture<Void> addConsumer(Consumer consumer) {

  if (getActiveConsumer() != null) {
      return actConsumer.cnx().checkConnectionLiveness().thenCompose(actConsumerStillAlive -> {
          if (actConsumerStillAlive.isEmpty() || actConsumerStillAlive.get()) {
              return FutureUtil.failedFuture(new ConsumerBusyException("Exclusive consumer is already"
                      + " connected"));
          } else {
              return addConsumer(consumer);
          }
      });
  }
}

Affects

It leads to the following subscribing can never succeed due to the ServerCnx maintaining a failed future and it can never be removed.

Error logs

2024-07-25T09:28:00,109+0000 [pulsar-io-4-30] WARN  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.2:48345][persistent://{tenant}/{namespace}/__change_events-partition-0][multiTopicsReader-381f38cdef] A failed consumer with id is already present on the connection, consumerId=7425
2024-07-25T09:28:00,109+0000 [pulsar-io-4-30] ERROR org.apache.pulsar.broker.service.ServerCnx - A failed consumer with id is already present on the connection. consumerId: 7425, remoteAddress: /127.0.0.2:48345, subscription: multiTopicsReader-381f38cdef
...
...
...
// loop

Modifications

Fix the bug.

Documentation

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

Matching PR in forked repository

PR in forked repository: x

[github-actions]

@poorbarcode Please add the following content to your PR description and select a checkbox:

- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

[RobertIndie]

LGTM

[poorbarcode]

/pulsarbot rerun-failure-checks

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 69.23077% with 4 lines in your changes missing coverage. Please review.

Project coverage is 74.54%. Comparing base (bbc6224) to head (ae26a3e).
Report is 494 commits behind head on master.

Files	Patch %	Lines
...va/org/apache/pulsar/broker/service/ServerCnx.java	69.23%	3 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #23123      +/-   ##
============================================
+ Coverage     73.57%   74.54%   +0.96%     
- Complexity    32624    34063    +1439     
============================================
  Files          1877     1919      +42     
  Lines        139502   144143    +4641     
  Branches      15299    15755     +456     
============================================
+ Hits         102638   107446    +4808     
+ Misses        28908    28485     -423     
- Partials       7956     8212     +256     

Flag	Coverage Δ
inttests	27.83% <30.76%> (+3.25%)	⬆️
systests	24.66% <0.00%> (+0.34%)	⬆️
unittests	73.89% <69.23%> (+1.04%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files	Coverage Δ
...va/org/apache/pulsar/broker/service/ServerCnx.java	72.76% <69.23%> (+0.62%)	⬆️

... and 484 files with indirect coverage changes

[lhotari]

LGTM