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

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`:

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

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

```java
// 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:

```java
// 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:

```java
// 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:

```java
// 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:

```java
// 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.