[improve][broker] Add a broker-level memory limit for in-flight replication

[poorbarcode]

Motivation

When many topics replicate concurrently, the total memory is unbounded and unpredictable, easily causing OOM.

Modifications

Add a broker-level memory limit for in-flight replication

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

[lhotari]

What is a reason to introduce a deprecated method in new code?

[Dream95]

During local debugging, acquiredBytes is 0, which hits the No memory budget available, waiting path and causes the test to fail.

[lhotari]

Thanks for working on this — bounding replication memory is a real concern, and the per-replicator hole you're pointing at does exist. I ran a local review of this PR with Claude (Fable) and Codex (gpt-5.6-sol) and then verified each finding against the code myself; the analysis below comes from that review, and I take responsibility for what I'm passing on here.

The headline is that I think the motivation needs re-checking against current master before the implementation is worth iterating on, because I believe the window this PR targets is already covered. Separately, there are four issues that would block merging as-is.

The premise: InflightReadsLimiter already covers this window, and it's on by default

The PR description says in-flight replication memory is unbounded broker-wide. As far as I can tell that isn't true on master anymore:

1. The existing permit is released on entry deallocation, not on read completion. RangeEntryCacheImpl#asyncReadEntriesByPosition wraps the read callback and registers an onDeallocate hook per entry, releasing the handle only when the last entry of the batch drops to refcount 0:

// RangeEntryCacheImpl.java:357-372
public void readEntriesComplete(List<Entry> entries, Object ctx2) {
    if (!entries.isEmpty()) {
        // release permits only when entries have been handled
        AtomicInteger remainingCount = new AtomicInteger(entries.size());
        for (Entry entry : entries) {
            ((EntryImpl) entry).onDeallocate(() -> {
                if (remainingCount.decrementAndGet() <= 0) {
                    pendingReadsLimiter.release(handle);
                }
            });
        }
    } else {
        pendingReadsLimiter.release(handle);
    }

For a replicator, that last entry.release() happens in ProducerSendCallback#sendComplete (PersistentReplicator.java:562) — i.e. after the remote cluster's SendReceipt. That is the same window this PR accounts for, released one line after the PR's own release(...) call at :561. The config doc states the intent by name: "the memory retained by data read from storage (or cache) until it has been delivered to the Consumer Netty channel" (ServiceConfiguration.java:2542-2546).

2. Replicator reads do flow through it. PersistentReplicator.java:365 → ManagedLedgerImpl.java:2468 → RangeEntryCacheImpl.java:268-271, which hardcodes acquirePermits = true. Permits are acquired before the cache lookup (:312-337), so cache hits are charged too.

3. It's enabled by default. ManagedLedgerClientFactory.java:99-104 — when managedLedgerMaxReadsInFlightSizeInMB is unset it resolves to max(15% of JVM max direct memory, dispatcherMaxReadSizeBytes). Only an explicit 0 disables it.

If that reading is right, this PR would double-charge every replicated read against two independently-estimated budgets covering near-identical windows, with no guidance on how to size them relative to each other.

The existing limiter also handles the hard parts this PR has to re-solve: an estimate floor of max(observedAvg, 10 KiB) when a ledger has no stats yet (RangeEntryCacheImpl.java:508-514), release on readEntriesFailed (:376-379), a bounded acquire queue with a 60 s timeout, a guarantee that a single oversized read is always admitted when the budget is free (InflightReadsLimiter.java:194-204), and OTel/Prometheus metrics.

Two cases where the premise would still hold, and which I think are the right things to chase instead:

• managedLedgerCacheSizeMB=0 returns EntryCacheDisabled (RangeEntryCacheManagerImpl.java:86-89), which never calls the limiter at all — reads are entirely unaccounted. Worth fixing on its own, for all read paths.
• replicationProducerQueueSize (default 1000) feeds .maxPendingMessages(...) (AbstractReplicator.java:158) and getInflightMessagesCount() sums entry counts only (PersistentReplicator.java:1032-1046). That's a genuine per-replicator byte hole — up to ~5 GB per replicator at the max message size — and it's count-based, not byte-based.

Blockers in the current implementation

B1 — Replication of new topics stalls with the feature disabled (the shipped default).
getAverageEntrySize() (PersistentReplicator.java:373-389) reads LedgerInfo from getLedgersInfo(), but the currently-open ledger is stored as new LedgerInfo().setLedgerId(...).setTimestamp(0) with entries/size populated only at close (ManagedLedgerImpl.java:654 vs :1969-1980). On a topic still in its first ledger, lowerEntry() is null, so it returns 0. Then tryAcquiredBytes = readLimitOnMsg * 0 = 0, and acquireUpTo(0, ...) returns requestedBytes = 0 on the maxBytes <= 0 short-circuit (BrokerReplicationMemoryLimiter.java:61-63) → ReadLimits(0, 0, 0) → not readable → one futile retry per second (MESSAGE_RATE_BACKOFF_MS). Replication does not progress until the first ledger rolls. The same applies to any topic whose closed ledgers have been trimmed. The sentinel 0 == "no budget" collides with "requested 0". The CI failures on this PR (Broker Groups 1/2/4/5 and Integration-Messaging) look consistent with this.

B2 — Every filtered entry leaks its bytes permanently.
readEntriesComplete calibrates the budget to the actual total of all read entries (PersistentReplicator.java:486-487), but release(entry.getLength()) has exactly one call site (:561), reached only by entries that make it to producer.sendAsync. Every skip branch in replicateEntries returns the entry without touching the limiter — GeoPersistentReplicator.java:164, 180, 187, 197, 211, 223, 236, 249, 272, plus the equivalents in ShadowReplicator. Line 211 is the msg.isReplicated() branch, which is the steady-state reverse direction of any bidirectional geo-replication setup — roughly half the traffic. Once enabled, inflightBytes ratchets monotonically to the cap and acquireUpTo then returns 0 for every replicator on the broker, permanently, with the buffers long since freed. Only a config change or restart recovers.

B3 — readEntriesFailed and cancelled reads never return the estimate.
completeFailedReadTask (PersistentReplicator.java:678-694) sets empty entries and returns with no limiter call, so each failed read leaks up to readBatchSize × avgEntrySize — under bookie throttling, i.e. exactly the overload this is meant to mitigate. Same for cancelled pending reads: cancelPendingReadTasks (:1096-1101) notes in its own comment that "the task will never receive a read completed callback if cancel pending reading successfully", so the calibrate(estimatedBytes, 0) refund at :415 never runs. That path is hit by routine topic unload/terminate, failed-publish rewind, and the schema-fetch pause.

B4 — The counter drifts negative while disabled, so the feature can't be turned on later; and broker.conf is ignored.
With maxBytes == 0, acquireUpTo returns early without incrementing (BrokerReplicationMemoryLimiter.java:61-63), but calibrate and release decrement unconditionally from unconditional callers (PersistentReplicator.java:415, :487, :561). Net −estimate per read cycle on the default config, so a later dynamic enable starts from a large negative baseline and limits nothing. Compounding this: the limiter is constructed with maxBytes = 0 (BrokerService.java:390) and the only writer is the dynamic listener (:3313-3315) — registerConfigurationListener (:3487-3490) only stores the callback and fires on dynamic change, and nothing reads getReplicationMaxInflightMemorySizeMB() at startup. So setting it in broker.conf silently does nothing, and dynamic config is the only way in.

Smaller issues

• pendingTasks is an unbounded ConcurrentLinkedQueue drained one waiter per release, while PersistentReplicator.java:339 re-offers a fresh this::readMoreEntriesAsync on every 1 s poll with no dedup. During the stall B2/B3 produce there are no releases at all, so it grows ~1 task/sec/replicator indefinitely, each holding a strong reference to its replicator and topic. setMaxBytes also only drains when the new value is > 0 (:118-123), so dynamically disabling strands whatever is queued.
• The limiter's executor is workerGroup, the client-facing Netty IO group (BrokerService.java:390), and readMoreEntriesAsync then re-dispatches to brokerService.executor() — which is workerGroup (BrokerService.java:3085-3087). A pointless hop, and execute can throw RejectedExecutionException out of release() during shutdown.
• Acquired bytes don't cap the read: :289/:365 still pass readLimitOnByte, so a read can overshoot the remaining budget and calibrate then pushes past maxBytes.
• No metrics at all — the only signal that all replication has stalled is a log.debug (:282). The existing limiter exports four OTel instruments plus Prometheus gauges.
• tryAcquire is @Deprecated from birth (BrokerReplicationMemoryLimiter.java:107-111); releaseEstimated, isLimitReached, PendingTask.estimatedBytes and PersistentReplicator#estimateBytes (:391-396) are unused outside the new unit test. Worth deleting.
• java.util.Map.Entry is fully qualified at PersistentReplicator.java:384, against the import guidance in CODING.md.
• Test coverage: the new unit test exercises the limiter in isolation with correctly paired acquire/release calls — which is the exact invariant the integration code doesn't maintain. None of the leak paths (read failure, cancelled read, filtered entries, failed publish, schema pause), the startup-config path, or the disable→enable transition are covered, and there's no end-to-end test with a real replicator.

Suggested direction

I'd suggest not iterating on this shape. All four blockers come from one choice: manual acquire/release pairing on an AtomicLong, admitted on an estimate, instead of binding release to EntryImpl.onDeallocate the way RangeEntryCacheImpl.java:357-372 does. That contract is leak-proof by construction — a newly added continue branch can't defeat it — whereas pre-admission on an estimate is what forces calibrate(), and calibrate() is what produces the leak, the drift, and the stall.

Ranked alternatives, roughly by effort:

1. Make the replicator's own bound byte-aware — a replicationProducerQueueSizeBytes alongside the existing count config, clamping getPermitsIfNoPendingRead() (PersistentReplicator.java:1019-1030). InFlightTask already holds the entries, so per-task byte occupancy is derivable with no global counter, no estimate and no calibration. This targets the quantity that actually is unbounded.
2. Make managedLedgerMaxReadsInFlightSizeInMB dynamic (volatile + setter + dynamic = true) — a few lines, and it closes the only capability this PR's new config genuinely adds over the status quo.
3. Add limiter calls to EntryCacheDisabled, closing the managedLedgerCacheSizeMB=0 bypass for all read paths, not just replication.
4. A second InflightReadsLimiter instance selected by class of service, if isolating replication from consumer dispatch is the real goal — a field in RangeEntryCacheManagerImpl and selection at the single accessor RangeEntryCacheImpl.java:132-133. It inherits the onDeallocate release contract, bounded queue, timeout, oversize clamp and metrics. Needs some metrics work (the Prometheus gauges are static final and set absolutely) and a PIP.

Questions

1. What was the motivating incident? If it was direct-memory OOM from replication reads, was managedLedgerMaxReadsInFlightSizeInMB enabled at the time, and was the entry cache enabled? If it was consumer dispatch being starved by a stalled WAN replicator, that's a genuine gap that the existing single shared budget can't express — and it's the case that would justify option 4.
2. The branch is named pip-479-impl, but I don't see a pip/pip-479.md in-tree and the PR doesn't link one. A new user-facing broker config plus a broker-wide limiter should have a PIP on dev@ first, and that discussion is where the overlap with managedLedgerMaxReadsInFlightSizeInMB would surface.
3. The PR template checkboxes are all unchecked despite adding a new configuration — worth ticking the config/deployment item and highlighting the change.

Caveat on everything above: it's a static reading of the code — I haven't reproduced B1–B4 with a test. Happy to be shown where I've misread something, particularly on the onDeallocate release timing, since that's what the rest of the argument rests on.