[Improve] `Consumer.resume()` can leave the consumer stuck and not deliver any messages

[PavelZeger]

Search before reporting

• I searched in the issues and found nothing similar.

Motivation

I ran into this while adding pause/resume to the Go client (apache/pulsar-client-go#1507). The Go implementation mirrors the Java one, and during review I found the same gap exists in the Java client too. I already fixed it in the Go PR (apache/pulsar-client-go#1507); this issue is to bring the same fix back to Java.
When you pause a consumer and later resume it, resume() is supposed to tell the broker "ok, start sending me messages again." It does this by calling increaseAvailablePermits(cnx(), 0):

// ConsumerImpl.resume()
public void resume() {
    if (paused) {
        paused = false;
        increaseAvailablePermits(cnx(), 0);
    }
}

The problem is that increaseAvailablePermits only actually sends permits once the owed count reaches half the receiver queue size:

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

That "half the queue" check is fine for normal running — you don't want to send a flow command for every single message. But it's the wrong thing to do on resume. If, right when you resume, the consumer owes fewer permits than that and the broker has no permits left, then resume sends nothing. And since no messages are coming in, the owed count never goes up, so a flow command is never sent. The consumer just sits there quietly, even though you resumed it. This isn't slow - it's stuck. There's no way for it to recover on its own. When does this happen? Mostly when not every delivered message bumps the permit count by one (for example, duplicate or chunk fragments that get thrown away). In that case the owed count can end up below the threshold while the broker has already sent everything it was allowed to. Small receiverQueueSize values make it easier to hit. In the normal case the count lands back on a full queue size, so it usually works - which is why nobody noticed and the current tests don't catch it.

Solution

Give resume its own way to send permits that ignores the "half the queue" check and just sends whatever is owed. Since it sends exactly what's owed and nothing more, it can never send too much. This is the same fix I used in the Go PR.

@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);
        }
    }
}

It uses the same compare-and-swap trick the existing code already uses, so it won't race or send twice. If the consumer happens to be disconnected, sendFlowPermitsToBroker does nothing and the reconnect path grants a fresh batch anyway, so that case is safe. And MultiTopicsConsumerImpl.resume() just calls resume on each child, so this one change covers partitioned and multi-topic consumers too.

For a test: the current testPauseAndResume drains the whole queue, so the owed count ends up back at a full queue size and the bug never shows. A real regression 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. A small unit test in ConsumerImplTest with a mocked connection is the easiest way to do that reliably.

Alternatives

No response

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[lhotari]

But it's the wrong thing to do on resume. If, right when you resume, the consumer owes fewer permits than that and the broker has no permits left, then resume sends nothing.

I don't think that it's completely wrong and I don't see directly why the existing behavior would be causing the consumer to get stuck. However I do agree that it would make sense to send flow permits for all available permits after resuming the consumer.

When pause is called, the available permits should eventually get in the state where the consumer has available permits >= receiver queue size / 2. There must be other underlying bugs if the current logic doesn't resume the consumer.

When does this happen? Mostly when not every delivered message bumps the permit count by one (for example, duplicate or chunk fragments that get thrown away). In that case the owed count can end up below the threshold while the broker has already sent everything it was allowed to.

It seems that there are existing bugs if this happens. The permits handling should be only about what has been sent or received on the wire. If the broker sends duplicates, it shouldn't matter. Permit leaks would also impact other use cases, not just pause/resume. That's why finding and fixing the root cause is important.

In permit accounting the invariant is brokerInFlight + queuedAtClient + availablePermits == receiverQueueSize. A consumer is only truly stuck when the broker has 0 permits, the client queue is empty, and available permits < receiverQueueSize / 2. If the invariant holds, broker-idle + empty-queue forces available == receiverQueueSize, so resume would flush. A real stall therefore requires that this invariant is broken.

[PavelZeger]

You're right @lhotari , and I'll fix my framing.

The examples I gave don't actually leak in Java - duplicates, skipped batch entries, and intermediate
chunks all return their permit. So with the invariant holding, today's resume() would already flush. And I can't reproduce an actual stuck consumer, so I'll drop that claim.

I also can't really justify it on over-granting: Java's resume() already calls increaseAvailablePermits(cnx, 0), which sends only what's owed, not a full batch - so it can't over-grant today either. The only real difference my change can make is when owed < receiverQueueSize/2 on resume: current code waits, mine flushes right away. Without a leak that's not a stall, just a small thing - it tops the broker's window back to full the moment you resume instead of letting it refill as messages get consumed.

So I'm happy either way:

• if you think it's not worth it, I'll close this;
• if you think the eager top-up on resume earns its lines, I'll keep it minimal - the small flushAvailablePermitsToBroker change plus a ConsumerImplTest unit test, framed as "resume tops up the window" rather than a bug fix.

WDYT?