apache / apache/pulsar

[Bug] ProducerImpl releases a message's memory reservation twice in the terminal-state branch, permanently degrading the client memory limit

Open
#26,469 0 comments 0 reactions 1 assignee Claimed by @lhotari View on GitHub
type/bug
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 14h
Merged PRs (30d)
160

Description

### Search before reporting

- [x] I searched the [issues](https://github.com/apache/pulsar/issues) and found nothing similar. Related but distinct: #26343 and #26344 both concern producer admission control, and neither covers this accounting bug.

### Read release policy

- [x] I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

### User environment

Affects the Java client on **master** and **branch-4.2**, i.e. released versions **4.2.0** and later (`git tag --contains 1b5c818a538e` lists `v4.2.0` and every `v4.2.0.x` since). Not present on branch-4.0 / branch-4.1 / branch-3.x.

### Issue Description

`ProducerImpl.processOpSendMsg` releases a message's memory reservation **twice** in its terminal-state branch.

`releaseSemaphoreForSendOp` already returns the memory:

https://github.com/apache/pulsar/blob/master/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L1439-L1444

```java
protected void releaseSemaphoreForSendOp(OpSendMsg op) {
semaphoreRelease(isBatchMessagingEnabled() ? op.numMessagesInBatch : 1);
client.getMemoryLimitController().releaseMemory(op.uncompressedSize);
}
```

but the terminal-state branch calls it and then releases the same bytes again:

```java
final State state = getState();
if (state == State.Terminated || state == State.Closed || state == State.ProducerFenced) {
// The producer is in a terminal state and will never reconnect. Fail the message immediately
// rather than leaving it stuck in pendingMessages until sendTimeout.
releaseSemaphoreForSendOp(op);
client.getMemoryLimitController().releaseMemory(op.uncompressedSize); // <-- second release
op.sendComplete(getTerminalException(state));
...
}
```

It is the only one of the four `releaseSemaphoreForSendOp` call sites in the class that adds an explicit `releaseMemory` on top; the other three (`ackReceived`, the checksum-failure path, the not-allowed path) just call the helper.

**Why it does not simply self-correct.** `MemoryLimitController.currentUsage` is a plain `AtomicLong` with no floor, and `releaseMemory` subtracts unconditionally:

```java
long newUsage = currentUsage.addAndGet(-size);
```

while `tryReserveMemory` rejects only while `current > memoryLimit`:

```java
if (current > memoryLimit && memoryLimit > 0) {
return false;
}
```

So every trip through this branch permanently lowers the counter by `op.uncompressedSize`. The client's memory limit is not restored by anything — there is no re-baselining — so it degrades monotonically, and once the drift exceeds the configured limit the limiter can no longer reject anything at all. At that point `memoryLimit` is silently inoperative for the whole client: `blockIfQueueFull(true)` stops blocking and `blockIfQueueFull(false)` stops failing, and a producer can buffer without bound until the heap is gone.

That matters more than it used to, because for a producer with no `maxPendingMessages` the client memory limit is the only backpressure there is. It is the *only* backpressure for the V5 client, which exposes no pending-message knob at all (see the note in `PulsarClientImpl#applyNoMemoryLimitProducerDefaults`).

**Reachability.** `isValidProducerState` rejects `Terminated` / `Closed` / `ProducerFenced` *before* `canEnqueueRequest`, so a fresh `sendAsync` never reaches this branch. It needs a message that was admitted while the producer was `Ready`/`Connecting` and whose state turns terminal before `processOpSendMsg` runs — a topic termination, a producer fencing, or a close racing with an in-flight send. So this leaks per racing message rather than on every send, which is also why it is easy to miss.

**A second, likely instance of the same shape**, in the `catch (Throwable)` at the end of the same method:

```java
} catch (Throwable t) {
releaseSemaphoreForSendOp(op);
...
}
```

The `try` block adds the op to `pendingMessages` before several of the statements that can throw, so an op released here can still be released again later by `ackReceived` or `failPendingMessages`. I have not constructed a case that proves this one, so treat it as worth checking rather than as established.

### Error messages / stacktraces

None — this fails silently. There is no log line and no exception; the only symptom is a memory limit that has quietly stopped limiting, which surfaces later as unbounded producer buffering or an OOM.

### Reproducing the issue

I found this by inspection while investigating producer backpressure, and I could not build a deterministic reproducer: the window between `isValidProducerState` and `processOpSendMsg` is narrow and I could not close it without a test hook. A unit test that flips the producer's state between admission and dispatch (a Mockito spy on the state accessor, or a `@VisibleForTesting` seam) and then asserts `memoryLimitController.currentUsage() == 0` should pin it — the existing cases in `ProducerMemoryLimitTest` already assert exactly that invariant after other failure paths, so the assertion style is established.

The duplicate release itself needs no reproducer to see; it is visible in the source, and the surrounding call sites establish the intended contract.

### Additional information

Introduced by `1b5c818a538e`, "[fix][client] Fail messages immediately in ProducerImpl when in terminal state" (#25317).

The fix is to delete the second `releaseMemory` call. Two things worth doing alongside it:

1. A test covering the terminal-state branch's accounting, as above.
2. Consider flooring `MemoryLimitController.currentUsage` at zero, or asserting non-negative in tests. It would not make an over-release correct, but it would turn "the limit silently stops working for the rest of the client's life" into a bounded, self-healing error — a much better failure mode for an invariant enforced by hand across a dozen call sites.

### Are you willing to submit a PR?

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

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.