Fixed race condition with NPE on Producer.closeAsync() (#490)

pulsar-client/src/main/java/com/yahoo/pulsar/client/impl/ProducerImpl.java

@@ -406,7 +406,8 @@ public class ProducerImpl extends ProducerBase implements TimerTask {
 
         stats.cancelStatsTimeout();
 
-        if (getClientCnx() == null || currentState != State.Ready) {
+        ClientCnx cnx = cnx();
+        if (cnx == null || currentState != State.Ready) {
             log.info("[{}] [{}] Closed Producer (not connected)", topic, producerName);
             synchronized (this) {
                 setState(State.Closed);
@@ -428,7 +429,6 @@ public class ProducerImpl extends ProducerBase implements TimerTask {
         ByteBuf cmd = Commands.newCloseProducer(producerId, requestId);
 
         CompletableFuture<Void> closeFuture = new CompletableFuture<>();
-        ClientCnx cnx = cnx();
         cnx.sendRequestWithId(cmd, requestId).handle((v, exception) -> {
             cnx.removeProducer(producerId);
             if (exception == null || !cnx.ctx().channel().isActive()) {