[java-pubsub] Publisher.shutdown() can block forever: messages cancelled in the failure callback are never returned to messagesWaiter

[laughingman7743]

Version: google-cloud-pubsub 1.152.0. The line numbers below are from that release's sources,
and Publisher.java and Waiter.java are byte-identical at main today, so they apply there
too.

Summary

When a publish for an ordering key fails while further messages for that key are still accumulating
in the publisher's un-flushed batch, Publisher's pending-message counter is left permanently above
zero. Publisher.shutdown() waits on that counter, uninterruptibly and without a timeout, so it
never returns.

Detail

Publisher.publish counts every accepted message into messagesWaiter:

// Publisher.java:343
messagesWaiter.incrementPendingCount(1);

Both batch callbacks decrement by the size of the batch that was in flight:

// Publisher.java:567 (and 543 on the success path)
} finally {
  messagesWaiter.incrementPendingCount(-outstandingBatch.size());
}

Before that finally, the failure callback also cancels the messages that are still accumulating in
the un-flushed MessagesBatch for the failed ordering key, and drops the batch:

// Publisher.java:551-560
MessagesBatch messagesBatch = messagesBatches.get(outstandingBatch.orderingKey);
if (messagesBatch != null) {
  for (OutstandingPublish outstanding : messagesBatch.messages) {
    outstanding.publishResult.setException(
        SequentialExecutorService.CallbackExecutor.CANCELLATION_EXCEPTION);
  }
  messagesBatches.remove(outstandingBatch.orderingKey);
}

Each of those messages incremented the waiter when it was published, and none of them is part of any
OutstandingBatch, so nothing ever decrements for them. pendingCount stays at least equal to the
number of cancelled messages for the life of the publisher.

(The other paused-key path is balanced: a publish() for an already-paused key returns at
Publisher.java:314-317, before the increment at line 343.)

Why that hangs shutdown

Waiter.waitComplete() ignores interruption and wakes only on an exact zero:

// Waiter.java:33-56
public synchronized void incrementPendingCount(int delta) {
  this.pendingCount += delta;
  if (pendingCount == 0) {
    notifyAll();
  }
}

public synchronized void waitComplete() {
  boolean interrupted = false;
  try {
    while (pendingCount > 0) {
      try {
        wait();
      } catch (InterruptedException e) {
        // Ignored, uninterruptibly.
        interrupted = true;
      }
    }
  } finally { ... }
}

and shutdown() calls it directly:

// Publisher.java:674-682
public void shutdown() {
  ...
  publishAllOutstanding();
  messagesWaiter.waitComplete();
  backgroundResources.shutdown();
}

So the caller's thread parks permanently, and awaitTermination(timeout, unit) — the API that does
take a bound — is never reached, because it is documented and intended to be called after
shutdown().

Reproduction

Ordering enabled, an element-count threshold of 2, and a long delay threshold, against a topic that
does not exist:

// m1 + m2 meet the threshold and go out as one request; m3 stays in the un-flushed
// MessagesBatch for the same key while that request is in flight.
publish(publisher, "m1", "k");
publish(publisher, "m2", "k");
publish(publisher, "m3", "k");
// ... let the futures settle (NOT_FOUND), then:
publisher.shutdown();   // does not return

Run against the Pub/Sub emulator, shutdown() had not returned after 10 s. The same setup with the
third message removed, and the same setup with no ordering key, both returned immediately:

variant	shutdown() returned
3 keyed messages, one left batched behind the failing request	no
2 keyed messages, nothing left batched	yes
3 messages, no ordering key	yes

So the leftover message for the failing key is the variable, not the NOT_FOUND and not the
batching.

Suggested fixes

Either would resolve it; the first is the actual accounting bug.

1. Decrement the waiter for the messages the failure callback cancels, next to where it cancels
   them — the count removed from messagesBatches is exactly what is owed back.
2. Give shutdown() a bounded wait. Waiter already has tryWait(timeoutMillis, clock), and both
   MessageDispatcher (MessageDispatcher.java:360) and StreamingSubscriberConnection
   (StreamingSubscriberConnection.java:263) use it; Publisher.shutdown() is the only caller of
   the unbounded waitComplete().

Impact

Any application that calls shutdown() from a thread that must stay responsive — a request
handler, a scheduler, a lifecycle callback with its own timeout. awaitTermination(timeout, unit)
is the API that takes a bound, but it is documented to be called after shutdown(), so it is
never reached and the bound never applies. The calling thread is then unrecoverable short of
killing the process.