Skip to content

[fix][client] Release the reserved memory only once when failing a send in a terminal state - #26476

Open
SongOf wants to merge 1 commit into
apache:masterfrom
SongOf:fix-producer-terminal-state-double-memory-release
Open

[fix][client] Release the reserved memory only once when failing a send in a terminal state#26476
SongOf wants to merge 1 commit into
apache:masterfrom
SongOf:fix-producer-terminal-state-double-memory-release

Conversation

@SongOf

@SongOf SongOf commented Sep 6, 2026

Copy link
Copy Markdown
Contributor

Motivation

ProducerImpl#processOpSendMsg fails a message immediately when the producer is in a terminal
state (Terminated / Closed / ProducerFenced) instead of leaving it in pendingMessages
until the send timeout fires. That path releases the memory reserved for the message twice:

releaseSemaphoreForSendOp(op);
client.getMemoryLimitController().releaseMemory(op.uncompressedSize);

releaseSemaphoreForSendOp(op) already returns op.uncompressedSize to the
MemoryLimitController, so the explicit call right after it releases the same bytes a second
time.

MemoryLimitController only validates that the argument is non-negative, not that the running
total stays sane, so this fails silently: every message published after the producer reaches a
terminal state drives currentUsage further below zero. Once it is negative,
tryReserveMemory() never returns false again and memoryLimitBytes is effectively disabled
for the whole client — every producer and consumer sharing that PulsarClient loses its memory
limit, which is exactly the protection that is supposed to keep the client from running out of
memory. An application that keeps publishing to a terminated topic, or that publishes after the
producer was fenced, hits this on every send.

This is a regression from #25317, which introduced the terminal-state fast-fail branch.

Modifications

  • ProducerImpl#processOpSendMsg: drop the redundant
    client.getMemoryLimitController().releaseMemory(op.uncompressedSize) from the terminal-state
    branch, leaving the single release performed by releaseSemaphoreForSendOp(op). Added a short
    comment recording that releaseSemaphoreForSendOp() already gives the reserved memory back, so
    the call is not re-added later.

No other path changes: the semaphore permits, the send callback, the ByteBufPair release and
the OpSendMsg recycling in that branch are untouched.

Verifying this change

This change added tests and can be verified as follows:

  • ProducerImplTest#testProcessOpSendMsgInTerminalStateReleasesMemoryOnce — reserves a known
    number of bytes in a real MemoryLimitController, drives processOpSendMsg() once for each of
    the three terminal states (Terminated, Closed, ProducerFenced), and asserts that
    currentUsage() returns to exactly 0. On the unpatched code it fails with
    expected [0] but found [-128], i.e. for the real reason — the double release — rather than by
    forcing internal state. The OpSendMsg is built through the batched factory so that op.msg is
    null, which keeps the test on the memory-accounting path and out of the batch-flush and
    message-size checks.

Local run:

./gradlew :pulsar-client-original:test --tests "ProducerImplTest" -PtestRetryCount=0
BUILD SUCCESSFUL

./gradlew rat spotlessCheck checkstyleMain checkstyleTest
BUILD SUCCESSFUL

The whole :pulsar-client-original:test module also passes.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

  • Dependencies (add or upgrade a dependency)
  • The public API
  • The schema
  • The default values of configurations
  • The threading model
  • The binary protocol
  • The REST endpoints
  • The admin CLI options
  • The metrics
  • Anything that affects deployment

This is an internal bug fix in the client send path. It changes no public API, schema,
configuration default, wire protocol, REST endpoint, CLI option or metric; it restores the
intended behaviour of the existing memoryLimitBytes setting rather than changing it.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants