[improve][client] Add exception handle for client send error

[congbobo184]

Motivation

• producer send messages return error will close the current cnx
• if close the current cnx, all producers and consumers witch use this cnx will close and reconnect
  (

  pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java

  Lines 323 to 345 in 5c6602c

  	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  	super.channelInactive(ctx);
  	connectionsClosedCounter.increment();
  	lastDisconnectedTimestamp = System.currentTimeMillis();
  	log.info("{} Disconnected", ctx.channel());
  	if (!connectionFuture.isDone()) {
  	connectionFuture.completeExceptionally(new PulsarClientException("Connection already closed"));
  	}
  	ConnectException e = new ConnectException(
  	"Disconnected from server at " + ctx.channel().remoteAddress());
  	// Fail out all the pending ops
  	pendingRequests.forEach((key, future) -> {
  	if (pendingRequests.remove(key, future) && !future.isDone()) {
  	future.completeExceptionally(e);
  	}
  	});
  	waitingLookupRequests.forEach(pair -> pair.getRight().getRight().completeExceptionally(e));
  	// Notify all attached producers/consumers so they have a chance to reconnect
  	producers.forEach((id, producer) -> producer.connectionClosed(this, Optional.empty(), Optional.empty()));
  	consumers.forEach((id, consumer) -> consumer.connectionClosed(this, Optional.empty(), Optional.empty()));

  )
• this will influence a lot of producers and consumers
• we only close this producer and reconnect enough, don't need to close this cnx

Modifications

receive send_error, close current producer connection, then it will reconnect automatically

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

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:

[BewareMyPower]

Fix code style

[BewareMyPower]

LGTM. But not sure why an extra blank line between the import clauses passes the checkstyle check.

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 73.46%. Comparing base (bbc6224) to head (60b69a4).
Report is 463 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #23038      +/-   ##
============================================
- Coverage     73.57%   73.46%   -0.12%     
- Complexity    32624    33500     +876     
============================================
  Files          1877     1915      +38     
  Lines        139502   143963    +4461     
  Branches      15299    15730     +431     
============================================
+ Hits         102638   105759    +3121     
- Misses        28908    30094    +1186     
- Partials       7956     8110     +154     

Flag	Coverage Δ
inttests	27.78% <0.00%> (+3.19%)	⬆️
systests	24.73% <0.00%> (+0.40%)	⬆️
unittests	72.53% <100.00%> (-0.32%)	⬇️

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

Files	Coverage Δ
.../java/org/apache/pulsar/client/impl/ClientCnx.java	71.82% <100.00%> (+0.05%)	⬆️

... and 509 files with indirect coverage changes