[bartoszs321]

Summary

• keep failed single-topic consumer mailboxes draining requests so calls made after shutdown complete instead of hanging
• propagate multi-topic poller failures to the parent consumer and fail pending receives
• drain failed multi-topic mailbox requests and close child consumers when DisposeAsync is called
• add integration coverage for single-topic and multi-topic mailbox failures

Problem

When a consumer mailbox faulted, its reader stopped. Requests posted afterwards were accepted by the unbounded channel but never received a response. This could leave operations such as GetStats, ReceiveAsync, and DisposeAsync waiting indefinitely. A multi-topic poller failure was only logged, leaving the parent consumer apparently ready even though it could no longer deliver messages.

Implementation

After a mailbox exits, a lightweight drain continues reading its channel. Requests that can no longer be performed fail with AlreadyClosedException; close requests complete after the necessary local or child-consumer cleanup. Multi-topic poller failures are posted to the parent mailbox, which moves the consumer to a failed state and completes pending receives.

Verification

• Unit tests: 216 passed, 2 skipped
• Mailbox integration tests against Apache Pulsar 3.3.0: 3 passed

[Copilot]

Copilot review overview

🟡 Changes recommended

Faulted single-topic consumers remain in the wrong lifecycle state after disposal, and pending-receive coverage is incomplete.

Review effort: Balanced
Findings: 2

Open (2)

• Close request leaves disposed consumer in Failed state · New
• Test queued ReceiveAsync failure when poller stops · New

What changed in this PR

Improves consumer failure handling so mailbox failures no longer leave requests hanging.

Changes:

• Drains stopped consumer mailboxes and completes requests.
• Propagates multi-topic poller failures.
• Adds mailbox-failure integration coverage.

File	Description
ConsumerImpl.fs	Drains requests after mailbox termination.
MultiTopicsConsumerImpl.fs	Handles poller failures and child cleanup.
Basic.fs	Tests mailbox-failure behavior.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Lanayx]

This should be NotConnectedException

[bartoszs321]

Updated in 29edfc2

[Lanayx]

Thank you for the PR, would you be able to implement similar behavior for Producer, so it will be single PR for both

[Copilot]

Copilot review overview

🟡 Changes recommended

Stopped consumers can retain pooled payloads, and failed partition cleanup is incorrectly reported as successful.

Review effort: Balanced
Findings: 1

Open (1)

• DisposeAsync hides partition producer close failures · New

Resolved since last review (2)

• Test queued ReceiveAsync failure when poller stops
• Close request leaves disposed consumer in Failed state

Previously missed (1)

In code that hasn't changed since last review

Mailbox failure drops messages without disposing payloads

src/​Pulsar.Client/​Internal/​ConsumerImpl.fs:891

The catch-all also drops queued MessageReceived values without disposing their RawMessage.Payload. These payloads come from MemoryStreamManager, and every normal discard path explicitly disposes them (for example, ConsumerImpl.fs:1002-1052), so messages racing with a mailbox failure retain pooled buffers. Add a stopped-mailbox case that disposes the payload before discarding the message.

[Copilot]

Copilot review overview

🟡 Changes recommended

A partitioned producer request can still hang when that request triggers the mailbox failure, and the multi-topic test does not exercise waiter cleanup.

Review effort: Balanced
Findings: 1

Open (1)

• Drain leaves the faulting request hanging · New

Resolved since last review (1)

• DisposeAsync hides partition producer close failures

[Copilot]

🟡 Changes recommended

Drains can complete disposal before cleanup and remain permanently rooted after shutdown.

4 open findings

• Consumer drain never terminates after disposal · New
• Multi-topic consumer drain leaks after disposal · New
• Partitioned producer drain leaks after disposal · New
• Producer drain never terminates after disposal · New

1 resolved since last review

• Drain leaves the faulting request hanging

🧠 Review effort: Balanced

[bartoszs321]

The copilot review comments assume that a ReadAsync which never completes keeps the drain task, and everything it references, in memory forever. The .NET garbage collector doesn't work that way. A pending task is not a garbage collection root; it stays alive only while something reachable references it.

In this case the drain's waiting operation is held by the mailbox channel. The channel is held only by the consumer or producer that owns it, and the drain's closure points back to that same object. Once the object is disposed, nothing outside that group references it:

• The client removes it from its consumers or producers set.
• The connection (ClientCnx) removes its registration.
• Its timers are stopped.
• The connection handler's loop exits.

What's left is objects that only reference each other, and the garbage collector frees groups like that.

The suggested fix would also cause a regression. Completing the channel writer makes later TryWrite calls return false. postAndAsyncReply ignores that result, so every public method that still posts after a stop would hang. Those methods include GetStats and IsConnected on the multi-topic consumer and the partitioned producer, which is exactly the hang this PR fixes. Avoiding that would mean adding state checks to every public method and losing the agreed stopped-mailbox replies: IsConnected returning false and Close returning Ok.

This was tested empirically by faulting and disposing an instance of all the implementations and checking that they are all collected by the GC under a forced collection while the drain is still pending. Also verified that completing the writer causes postAndAsyncReply hangs because it ignores TryWrite

[Lanayx]

Astra review

Thanks for the updates—the in-flight request handling and regression tests address the earlier concern. I also agree that a pending ReadAsync does not, by itself, permanently root the object graph, so I don’t think the drain needs to be removed or the writer completed on that basis.

There is one remaining ordering issue: all four mailbox continuations call drainStoppedMailbox() before performing failure handling. The drain can therefore process a queued Close before cleanup finishes. For single producers/consumers, that can complete DisposeAsync before the local cleanup assumed by the stopped-mailbox handler has happened. For wrappers, the drain’s close handler can also race with the failure continuation’s state updates.

Could you start the drain after the failure-state transition, cleanup, and in-flight-request handling, while ensuring it still starts if cleanup throws—for example, using try/finally? Please also add a deterministic regression test that queues a close around mailbox failure and verifies it cannot complete before required local cleanup.

This should be a focused ordering fix, not a mailbox redesign. Once that is addressed and CI is green, I’m happy to approve.

[Copilot]

🔵 Needs a closer look

Reply continuations can still block mailbox drains, and consumer-close responses are not registered.

4 open findings

• Producer drain never terminates after disposal
• Partitioned producer drain leaks after disposal
• Multi-topic consumer drain leaks after disposal
• Consumer drain never terminates after disposal

Previously missed (4)

In code that hasn't changed since last review

Use TaskCreationOptions for asynchronously run reply continuations

src/​Pulsar.Client/​Internal/​ConsumerImpl.fs:907

This drain can still deadlock on a reentrant caller continuation. postAndAsyncReply constructs its TCS with TaskContinuationOptions.RunContinuationsAsynchronously (Common/Tools.fs:162-165), which F# binds to the object-state constructor, so completing a direct reply task here can run an ExecuteSynchronously continuation inline. If that continuation synchronously waits on another consumer call, the sole drain reader cannot process the new request. Create reply TCS instances with TaskCreationOptions.RunContinuationsAsynchronously so stopped-mailbox replies cannot be blocked this way.

This issue also appears on line 1447 of the same file.

Use TaskCreationOptions for asynchronously run reply continuations

src/​Pulsar.Client/​Internal/​MultiTopicsConsumerImpl.fs:697

This drain can still deadlock on a reentrant caller continuation. postAndAsyncReply constructs its TCS with TaskContinuationOptions.RunContinuationsAsynchronously (Common/Tools.fs:162-165), which F# binds to the object-state constructor, so completing a direct reply task here can run an ExecuteSynchronously continuation inline. If that continuation synchronously waits on another multi-topic consumer call, the sole drain reader cannot process the new request. Create reply TCS instances with TaskCreationOptions.RunContinuationsAsynchronously.

Use TaskCreationOptions for asynchronously run reply continuations

src/​Pulsar.Client/​Internal/​PartitionedProducerImpl.fs:183

This drain can still deadlock on a reentrant caller continuation. postAndAsyncReply constructs its TCS with TaskContinuationOptions.RunContinuationsAsynchronously (Common/Tools.fs:162-165), which F# binds to the object-state constructor, so completing a direct reply task here can run an ExecuteSynchronously continuation inline. If that continuation synchronously waits on another partitioned-producer call, the sole drain reader cannot process the new request. Create reply TCS instances with TaskCreationOptions.RunContinuationsAsynchronously.

Use TaskCreationOptions for asynchronously run reply continuations

src/​Pulsar.Client/​Internal/​ProducerImpl.fs:576

This drain can still deadlock on a reentrant caller continuation. postAndAsyncReply constructs its TCS with TaskContinuationOptions.RunContinuationsAsynchronously (Common/Tools.fs:162-165), which F# binds to the object-state constructor, so completing a direct reply task here can run an ExecuteSynchronously continuation inline. If that continuation synchronously waits on another producer call, the sole drain reader cannot process the new request. Create reply TCS instances with TaskCreationOptions.RunContinuationsAsynchronously.

🧠 Review effort: Balanced