fix(pubsub): count a published message while its batch lock is held

The failure callback decrements messagesWaiter for the messages it cancels
out of a MessagesBatch, but publish() incremented after releasing
messagesBatchLock. A message visible in the batch and not yet counted would
therefore be decremented for without ever having been counted, taking
pendingCount below zero and letting waitComplete() return early.

Incrementing while the lock is still held makes "in a MessagesBatch" and
"counted" one state. Lock ordering is messagesBatchLock -> Waiter monitor
here and nowhere the reverse, and incrementPendingCount never blocks. The
paused-key path still returns before the increment, as before.

Author: laughingman7743

java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java

@@ -322,6 +322,12 @@ public ApiFuture<String> publish(PubsubMessage message) {
       }
 
       batchesToSend = messagesBatch.add(outstandingPublish);
+      // Counted while messagesBatchLock is still held, so that "in a MessagesBatch" and "counted"
+      // are one state. The failure callback decrements for the messages it cancels out of a
+      // MessagesBatch, so one that is visible there but not yet counted would take pendingCount
+      // below zero. Lock ordering is messagesBatchLock -> Waiter monitor here and nowhere the
+      // reverse, and incrementPendingCount never blocks.
+      messagesWaiter.incrementPendingCount(1);
       if (!batchesToSend.isEmpty() && messagesBatch.isEmpty()) {
         messagesBatches.remove(orderingKey);
       }
@@ -340,8 +346,6 @@ public ApiFuture<String> publish(PubsubMessage message) {
       messagesBatchLock.unlock();
     }
 
-    messagesWaiter.incrementPendingCount(1);
-
     // For messages without ordering keys, it is okay to send batches without holding
     // messagesBatchLock.
     if (!batchesToSend.isEmpty() && orderingKey.isEmpty()) {