ISSUE-12216: When the producer is closing, it should wait for all pending write requests to persist

[sijie]

Original Issue: apache#12216

---

Status: Under Discussion
Author: lordcheng10
Pull Request: apache#12195
Mailing List discussion: https://lists.apache.org/thread.html/r8bfcb7ab28612d94d441ff5eadd996413346f0780b6f7b3484aaf7dc%40%3Cdev.pulsar.apache.org%3E

Motivation

In the implementation of the producer's close method, the pending write request is not waited for persistence。
For example, in the following example, after the producer is closed, the data is not successfully written to the server：

try {
PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://127.0.0.1:6650").build();
Producer<byte[]> producer = client.newProducer().topic("test1").create();
producer.sendAsync("hello".getBytes());
client.close();
} catch (PulsarClientException e) {
e.printStackTrace();
}
Thread.sleep(1000);

In addition, the implementation of the org.apache.pulsar.client.impl.ProducerImpl#closeAsync method does not conform to the description of the Javadoc for closeAsync:
https://github.com/apache/pulsar/blob/04604724dd89a4c8ff40b922e80c7045fedf112d/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Producer.java#L178-L186

Goal

Ensure that the pending messages are persisted to the server when the producer is closed。

API Changes

The method of change is as follows:
org.apache.pulsar.client.impl.ProducerImpl#closeAsync