apache / apache/pulsar

[improve][client] Make `Consumer.resume()` eagerly top up the broker's permit window

Open
#25,978 5 comments 0 reactions 0 assignees View on GitHub
type/enhancement
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 14h
Merged PRs (30d)
160

Description

### Search before reporting

- [x] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.

### Motivation

I was adding pause/resume to the Go client (apache/pulsar-client-go#1507), and while doing that I took a closer look at how Java's `Consumer.resume()` tells the broker to start sending messages again. I found a small thing worth improving, and it's the same shape as what I fixed on the Go side. When you resume, Java calls `increaseAvailablePermits(cnx(), 0)`:

```java
public void resume() {
if (paused) {
paused = false;
increaseAvailablePermits(cnx(), 0);
}
}
```

But that method only actually sends permits to the broker once the owed count reaches half the receiver queue:

```java
while (available >= getCurrentReceiverQueueSize() / 2 && !paused) {
...
sendFlowPermitsToBroker(currentCnx, available);
...
}
```

That "wait until half the queue" rule is good for normal running - you don't want to send a flow command for every single message. The thing is, it's also applied on resume. So if you resume at a moment when fewer than half a queue of permits are owed, resume sends nothing right then, and the broker's window only fills back up later as you keep consuming.

One correction to how I first described this: it is **not** a stuck/stall bug. As long as the permit invariant holds - `brokerInFlight + queuedAtClient + availablePermits == currentReceiverQueueSize` - the consumer always recovers on its own, and in Java the invariant does hold (duplicates, skipped batch entries, and intermediate chunks all give their permit back). I couldn't reproduce a stuck consumer, so I'm dropping that claim. What's left is a nice-to-have: let resume top the broker's window straight back up to the full queue size instead of waiting for the threshold. Thanks @lhotari for pointing out the wrong framing in my first version.

### Solution

Give resume its own little flush that skips the half-queue rule and just sends whatever is owed. Because it only ever sends what's owed, it can never send too much:

```java
@Override
public void resume() {
if (paused) {
paused = false;
flushAvailablePermitsToBroker(cnx());
}
}

private void flushAvailablePermitsToBroker(ClientCnx currentCnx) {
int available = AVAILABLE_PERMITS_UPDATER.get(this);
while (available > 0 && !paused) {
if (AVAILABLE_PERMITS_UPDATER.compareAndSet(this, available, 0)) {
sendFlowPermitsToBroker(currentCnx, available);
break;
} else {
available = AVAILABLE_PERMITS_UPDATER.get(this);
}
}
}
```

A few notes:

- It uses the same compare-and-swap the existing code already uses, so there's no race and it won't send twice.
- The `available > 0` check means it sends nothing when nothing is owed - including when an auto-scale-down has left `availablePermits` at zero or below.
- If the consumer happens to be disconnected, `sendFlowPermitsToBroker` does nothing and the reconnect path grants a fresh batch anyway, so that's safe.
- `MultiTopicsConsumerImpl.resume()` just calls resume on each child, so partitioned and multi-topic consumers are covered too.

This is the same fix I already used in the Go client (apache/pulsar-client-go#1507).

On the auto-scaled receiver queue (`autoScaledReceiverQueueSizeEnabled(true)`): I don't think it needs any special handling. The queue only grows and shrinks on the consume path (`expectMoreIncomingMessages` / `reduceCurrentReceiverQueueSize`), never through `resume()`. The flush sends exactly the owed permits, which by the invariant is always at most the current queue size, so it can't over-grant no matter what size scaling has it at. And the scaled-down case (where permits can go to zero or negative) is handled by the `available > 0` guard.

On tests: the current `testPauseAndResume` drains the whole queue, so the owed count ends up back at a full queue size and the threshold path already flushes - which is exactly why the difference never shows today. A proper test needs to leave fewer permits owed than half the queue while the broker is at zero, then resume and check that a flow command actually goes out with exactly the owed count. The cleanest way is a `ConsumerImplTest` unit test with a mocked connection, plus an `autoScaledReceiverQueueSizeEnabled(true)` variant (scaled up: flush equals what's owed, nothing extra; scaled down with `availablePermits <= 0`: nothing sent).

### Alternatives

_No response_

### Anything else?

_No response_

### Are you willing to submit a PR?

- [x] I'm willing to submit a PR!

Contributor guide

Open the contributing guide

Research direction

Start with ConsumerImpl's existing increaseAvailablePermits logic and ConsumerImplTest, especially testPauseAndResume. Add focused mocked-connection coverage for resuming with fewer than half a queue of permits owed, including auto-scaled receiver queues. Done means resume sends exactly the owed positive permits immediately and sends nothing when the owed count is zero or negative.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
backend-api-design, distributed-systems
Issue type
Feature
Difficulty
3/5
Estimated time
1-2 days
Activity status
Quiet
Clarity
Clearly specified
Newbie friendliness
64/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.