[ServerCnx] Only reply to client when completing producerFuture (#13949)

### Motivation

We should only send the error response to the client when the code is able to complete the `producerFuture`. This logic is described here:

https://github.com/apache/pulsar/blob/2285d02aa9957af7877b9d3d3c628a750d813ca7/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java#L1286-L1293

Edit: in a previous version of this motivation section, I attributed the current behavior to #12874. That PR did not introduce this behavior, though.

### Modifications

* Move the response to the client into a conditional block that only runs when this section of the code is able to complete the future.

Author: michaeljmarshall

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -1259,16 +1259,17 @@ protected void handleProducer(final CommandProducer cmdProducer) {
                             (BrokerServiceException.TopicBacklogQuotaExceededException) cause;
                     IllegalStateException illegalStateException = new IllegalStateException(tbqe);
                     BacklogQuota.RetentionPolicy retentionPolicy = tbqe.getRetentionPolicy();
-                    if (retentionPolicy == BacklogQuota.RetentionPolicy.producer_request_hold) {
-                        commandSender.sendErrorResponse(requestId,
-                                ServerError.ProducerBlockedQuotaExceededError,
-                                illegalStateException.getMessage());
-                    } else if (retentionPolicy == BacklogQuota.RetentionPolicy.producer_exception) {
-                        commandSender.sendErrorResponse(requestId,
-                                ServerError.ProducerBlockedQuotaExceededException,
-                                illegalStateException.getMessage());
+                    if (producerFuture.completeExceptionally(illegalStateException)) {
+                        if (retentionPolicy == BacklogQuota.RetentionPolicy.producer_request_hold) {
+                            commandSender.sendErrorResponse(requestId,
+                                    ServerError.ProducerBlockedQuotaExceededError,
+                                    illegalStateException.getMessage());
+                        } else if (retentionPolicy == BacklogQuota.RetentionPolicy.producer_exception) {
+                            commandSender.sendErrorResponse(requestId,
+                                    ServerError.ProducerBlockedQuotaExceededException,
+                                    illegalStateException.getMessage());
+                        }
                     }
-                    producerFuture.completeExceptionally(illegalStateException);
                     producers.remove(producerId, producerFuture);
                     return null;
                 }