[fix][client] Fix producer/consumer stop to reconnect or Pub/Sub due to IO thread race-condition

[poorbarcode]

Motivation

Issue 1: see the test testUnstableNetWorkWhenCreatingProducer

The producer fails to start if the network is unstable.

time	socket closed	start producer
1		get a pooled connection from the pool
2		before registering the producer to the connection
3	socket closed
4	reconnect all producers, which is empty now
5		register the producer to a closed connection
6		failed to send Command.newProducer to brokers
7		users receive an error io.netty.channel.StacklessClosedChannelException: null

---

**Issue 2**: see the test: `testSendAfterCreateProducerAsync` - Start a producer/consumer async - Call a sync method in the callback method of method `createAsync`, such as `producer.send`, and `consumer.receive`. - The IO thread encounters a deadlock

Modifications

• fix the issues above

Documentation

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

Matching PR in forked repository

PR in forked repository: x

[lhotari]

Great work on addressing a hard problem, @poorbarcode! I added a few minor review comments.

[lhotari]

LGTM, great work @poorbarcode.

[nodece]

Issue 1:

It looks the CompletableFuture.supplyAsync(() -> clientCnx, clientCnx.ctx().executor()); can fix this issue, but I don’t understand this fix way.

In the old code, Isn't this callback thread clientCnx.ctx().executor()? I didn't double-check this part of the code.

LGTM

Issue 2:

You call the sync method in the async callback thread, this is an incorrect way. I don't suggest that you switch the thread to complete the callback.

In the past, we have removed many sync calls in the Pulsar.

[lhotari]

You call the sync method in the async callback thread, this is an incorrect way. I don't suggest that you switch the thread to complete the callback.

@nodece please provide the exact code location that you are referring to. Thanks!

[lhotari]

It looks the CompletableFuture.supplyAsync(() -> clientCnx, clientCnx.ctx().executor()); can fix this issue, but I don’t understand this fix way.

Indeed, it would be more consistent to use thenComposeAsync to switch the executor for all code branches instead of just a single branch.

[nodece]

You call the sync method in the async callback thread, this is an incorrect way. I don't suggest that you switch the thread to complete the callback.

@nodece please provide the exact code location that you are referring to. Thanks!

        client.newProducer().topic("test").createAsync().thenApply(producer -> {
            producer.send("hello-1".getBytes(StandardCharsets.UTF_8)); // incorrect, which will block the pulsar client thread.
            producer.sendAsync("hello-1".getBytes(StandardCharsets.UTF_8)); // correct
        });

User should not call a sync method in the callback of the async method.

+1