[Fix][Client] Fix pending message not complete when closeAsync

[pengxiangrui127]

Motivation

In the current process of closeAsync() in ProducerImpl.java, the client will firstly set the producer to the Closing state, then stop timerTask of sendTimeout, and then send a http request to broker to close the producer. When the client receives a successful response from broker, it sets the producer to the Closed state, removes the producer from the producers in client, and complete the PendingMessage with AlreadyClosedException.
However, if the communication between client and broker is abnormal due to network or other reasons, the PendingMessage will not be cleared anymore. What's worse is that since the producer is already in the Closing state at this time, there is no chance to trigger the cleanup of PendingMessage by calling the closeAsync() again, which is obviously unreasonable.

Modifications

There is no need to wait for the broker's response to clean up PendingMessage. Once you decide to shut down the producer, there is no need to retain the PendingMessage in the producer. At this time, failing quickly is more valuable than waiting.

Verifying this change

• Make sure that the change passes the CI checks.

Add a test case to ensure closeAndClearPendingMessages() has been invoke when request broker error after closeAsync().

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

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

Matching PR in forked repository

PR in forked repository:

[lhotari]

There's a related issue #11780

[lhotari]

There's also #12195

[lhotari]

LGTM. This PR makes sense.

I initially thought that the intention was to also address issues #11780 and #12195 as well. There could be a separately improvement addressing the problem that closeAsync doesn't flush any pending messages before closing. Usually there's an expectation coming from java.io.OutputStream that when you call close, it will flush before closing. This doesn't happen with Producer's close.

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 74.32%. Comparing base (bbc6224) to head (d2e261e).
Report is 809 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #23761      +/-   ##
============================================
+ Coverage     73.57%   74.32%   +0.75%     
+ Complexity    32624    32146     -478     
============================================
  Files          1877     1838      -39     
  Lines        139502   143048    +3546     
  Branches      15299    16232     +933     
============================================
+ Hits         102638   106323    +3685     
+ Misses        28908    28355     -553     
- Partials       7956     8370     +414     

Flag	Coverage Δ
inttests	26.72% <100.00%> (+2.13%)	⬆️
systests	23.67% <100.00%> (-0.65%)	⬇️
unittests	73.71% <100.00%> (+0.86%)	⬆️

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

Files with missing lines	Coverage Δ
...va/org/apache/pulsar/client/impl/ProducerImpl.java	84.09% <100.00%> (+0.50%)	⬆️

... and 1005 files with indirect coverage changes