googleapis / googleapis/google-cloud-java

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

Open
#14,001 0 comments 0 reactions 1 assignee Claimed by @michaelpri10 View on GitHub
api: pubsub
Dominant language
Java
Stars
2.1k
Forks
1.2k
Avg merge
1d 23h
Merged PRs (30d)
154

Description

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.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.