Skip to content

Fix consumer hangs after mailbox failure #374

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Lanayx merged 9 commits into from
Oct 9, 2026
Merged

Fix consumer hangs after mailbox failure #374

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
7746009
fix: complete consumer requests after mailbox failure
bartoszs321 Oct 5, 2026
4aae9bb
Merge remote-tracking branch 'upstream/develop' into fix/consumer-mai…
bartoszs321 Oct 5, 2026
29edfc2
Fix exception types from consumers
bartoszs321 Oct 6, 2026
e268599
complete producer requests after mailbox failure
bartoszs321 Oct 6, 2026
642486e
Dispose queued MessageReceived payloads
bartoszs321 Oct 6, 2026
649626c
Accurately set channel exception when replying from stopped producer …
bartoszs321 Oct 6, 2026
0171353
Track current message in partitioned producer, so it can be failed & …
bartoszs321 Oct 6, 2026
20519ac
Align tracking the current message and failing its reply on fault acr…
bartoszs321 Oct 7, 2026
ab742a3
Start the stopped-mailbox drain only after failure cleanup
Lanayx Oct 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 80 additions & 15 deletions src/Pulsar.Client/Internal/ConsumerImpl.fs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -857,10 +857,66 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
}

let mb = Channel.CreateUnbounded<ConsumerMessage<'T>>(UnboundedChannelOptions(SingleReader = true, AllowSynchronousContinuations = true))

let replyFromStoppedMailbox (ex: exn) (msg: ConsumerMessage<'T>) =
match msg with
| Receive receiveCallback ->
receiveCallback.MessageChannel.TrySetException ex |> ignore
| BatchReceive receiveCallbacks ->
receiveCallbacks.MessagesChannel.TrySetException ex |> ignore
| Acknowledge (struct(_, _, Some (_, channel))) ->
channel.TrySetException ex |> ignore
| RedeliverAllUnacknowledged (Some channel) ->
channel.TrySetException ex |> ignore
| SeekAsync (_, channel) ->
channel.TrySetException ex |> ignore
| HasMessageAvailable channel ->
channel.TrySetException ex |> ignore
| GetStats channel ->
channel.TrySetException ex |> ignore
| ReconsumeLater (_, _, _, channel) ->
channel.TrySetException ex |> ignore
| ClearIncomingMessagesAndGetMessageNumber channel ->
channel.TrySetException ex |> ignore
| Unsubscribe channel ->
channel.TrySetException ex |> ignore
| Close channel ->
// the consumer is stopped before this runs, so there is nothing left to close
channel.TrySetResult() |> ignore
| CancelWaiter (struct (_, channel)) ->
channel.TrySetCanceled() |> ignore
| CancelBatchWaiter (struct (_, _, channel)) ->
channel.TrySetCanceled() |> ignore
| MessageReceived (rawMessage, _) ->
// these payloads come from the pool and are otherwise discarded with the message
rawMessage.Payload.Dispose()
| _ ->
()

let stoppedMailboxException () =
match connectionHandler.ConnectionState with
| Closing | Closed | Terminated -> AlreadyClosedException(prefix + " already closed") :> exn
| _ -> NotConnectedException(prefix + " not connected") :> exn

// requests posted after the mailbox exits would otherwise never be completed
let drainStoppedMailbox () =
backgroundTask {
while true do
let! msg = mb.Reader.ReadAsync()
Comment on lines +903 to +905
try
replyFromStoppedMailbox (stoppedMailboxException ()) msg
with ex ->
Log.Logger.LogWarning(ex, "{0} failed to reply to a request after the mailbox stopped", prefix)
} |> ignore

let mutable currentMessage : ConsumerMessage<'T> option = None

do (backgroundTask {
let mutable continueLoop = true
while continueLoop do
match! mb.Reader.ReadAsync() with
let! msg = mb.Reader.ReadAsync()
currentMessage <- Some msg
match msg with
| ConsumerMessage.ConnectionOpened ->

match connectionHandler.ConnectionState with
Expand Down Expand Up @@ -1384,20 +1440,29 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
Log.Logger.LogError("{0} can't unsubscribe since not connected", prefix)
NotConnectedException "Not connected to broker" |> channel.SetException
} :> Task).ContinueWith(fun t ->
if t.IsFaulted then
let (Flatten ex) = t.Exception
Log.Logger.LogCritical(ex, "{0} mailbox failure", prefix)
match connectionHandler.ConnectionState with
| Ready clientCnx ->
// release the consumer at the broker, otherwise it stays attached to the subscription
clientCnx.SendAndForget(Commands.newCloseConsumer consumerId (Generators.getNextRequestId()))
clientCnx.RemoveConsumer consumerId
| _ ->
()
connectionHandler.Failed()
stopConsumer()
else
Log.Logger.LogInformation("{0} mailbox has stopped normally", prefix))
try
if t.IsFaulted then
let (Flatten ex) = t.Exception
Log.Logger.LogCritical(ex, "{0} mailbox failure", prefix)
match connectionHandler.ConnectionState with
| Ready clientCnx ->
// release the consumer at the broker, otherwise it stays attached to the subscription
clientCnx.SendAndForget(Commands.newCloseConsumer consumerId (Generators.getNextRequestId()))
clientCnx.RemoveConsumer consumerId
| _ ->
()
connectionHandler.Failed()
try
currentMessage |> Option.iter (replyFromStoppedMailbox ex)
with currentEx ->
Log.Logger.LogWarning(currentEx, "{0} failed to fail the in-flight request after mailbox failure", prefix)
stopConsumer()
subscribeTsc.TrySetException(ex) |> ignore
else
Log.Logger.LogInformation("{0} mailbox has stopped normally", prefix)
finally
// only after cleanup, so a queued Close can't complete before the consumer has stopped
drainStoppedMailbox())
|> ignore

do startStatTimer()
Expand Down
108 changes: 99 additions & 9 deletions src/Pulsar.Client/Internal/MultiTopicsConsumerImpl.fs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ type internal BatchAddResponse<'T> =

type internal MultiTopicConsumerMessage<'T> =
| Init
| PollerFailed of exn
| Receive of ReceiveCallback<'T>
| BatchReceive of ReceiveCallbacks<'T>
| MessageReceived of ResultOrException<Message<'T>> * TaskCompletionSource<unit>
Expand Down Expand Up @@ -626,15 +627,86 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration
if t.IsFaulted then
let (Flatten ex) = t.Exception
Log.Logger.LogCritical(ex, "{0} poller failure", prefix)
post this.Mb (PollerFailed ex)
else
Log.Logger.LogInformation("{0} poller has stopped normally", prefix)
)

let mb = Channel.CreateUnbounded<MultiTopicConsumerMessage<'T>>(UnboundedChannelOptions(SingleReader = true, AllowSynchronousContinuations = true))

let replyFromStoppedMailbox (ex: exn) (msg: MultiTopicConsumerMessage<'T>) =
match msg with
| Receive receiveCallback ->
receiveCallback.MessageChannel.TrySetException ex |> ignore
| BatchReceive receiveCallbacks ->
receiveCallbacks.MessagesChannel.TrySetException ex |> ignore
| MessageReceived (_, pollerChannel) ->
pollerChannel.TrySetException ex |> ignore
| Acknowledge (channel, _, _)
| NegativeAcknowledge (channel, _)
| AcknowledgeCumulative (channel, _, _)
| Unsubscribe channel
| Seek (_, channel)
| SeekWithResolver (_, channel)
| ReconsumeLater (_, _, channel)
| ReconsumeLaterCumulative (_, _, channel) ->
channel.TrySetException ex |> ignore
| RedeliverAllUnacknowledged (Some channel) ->
channel.TrySetException ex |> ignore
| HasReachedEndOfTheTopic channel
| HasMessageAvailable channel ->
channel.TrySetException ex |> ignore
| GetStats channel ->
channel.TrySetException ex |> ignore
| LastDisconnectedTimestamp channel ->
channel.TrySetException ex |> ignore
| IsConnected channel ->
channel.TrySetResult false |> ignore
| CancelWaiter (struct (_, channel)) ->
channel.TrySetCanceled() |> ignore
| CancelBatchWaiter (struct (_, _, channel)) ->
channel.TrySetCanceled() |> ignore
| Close channel ->
// the child consumers outlive a failed mailbox, so they still have to be closed
backgroundTask {
try
let! _ =
consumers
|> Seq.map (fun (KeyValue(_, (consumer, _))) -> consumer.DisposeAsync().AsTask())
|> Task.WhenAll
()
with Flatten closeEx ->
Log.Logger.LogError(closeEx, "{0} could not close all child consumers properly", prefix)
this.ConnectionState <- Closed
channel.TrySetResult() |> ignore
} |> ignore
| _ ->
()

let stoppedMailboxException () =
match this.ConnectionState with
| Closing | Closed -> AlreadyClosedException(prefix + " already closed") :> exn
| _ -> NotConnectedException(prefix + " not connected") :> exn

// requests posted after the mailbox exits would otherwise never be completed
let drainStoppedMailbox () =
backgroundTask {
while true do
let! msg = mb.Reader.ReadAsync()
Comment on lines +693 to +695
try
replyFromStoppedMailbox (stoppedMailboxException ()) msg
with ex ->
Log.Logger.LogWarning(ex, "{0} failed to reply to a request after the mailbox stopped", prefix)
} |> ignore

let mutable currentMessage : MultiTopicConsumerMessage<'T> option = None

do (backgroundTask {
let mutable continueLoop = true
while continueLoop do
match! mb.Reader.ReadAsync() with
let! msg = mb.Reader.ReadAsync()
currentMessage <- Some msg
match msg with
| Init ->

Log.Logger.LogDebug("{0} Init", prefix)
Expand Down Expand Up @@ -669,6 +741,12 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration
if this.ConnectionState = Failed then
continueLoop <- false

| PollerFailed ex ->
Log.Logger.LogError(ex, "{0} stopping because its poller failed", prefix)
this.ConnectionState <- Failed
stopConsumer()
continueLoop <- false

| MessageReceived (message, pollerChannel) ->

let hasWaitingChannel = waiters.Count > 0
Expand Down Expand Up @@ -1022,13 +1100,22 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration
channel.SetException ex
continueLoop <- true
}:> Task).ContinueWith(fun t ->
if t.IsFaulted then
let (Flatten ex) = t.Exception
Log.Logger.LogCritical(ex, "{0} mailbox failure", prefix)
this.ConnectionState <- Failed
stopConsumer()
else
Log.Logger.LogInformation("{0} mailbox has stopped normally", prefix))
try
if t.IsFaulted then
let (Flatten ex) = t.Exception
Log.Logger.LogCritical(ex, "{0} mailbox failure", prefix)
this.ConnectionState <- Failed
try
currentMessage |> Option.iter (replyFromStoppedMailbox ex)
with currentEx ->
Log.Logger.LogWarning(currentEx, "{0} failed to fail the in-flight request after mailbox failure", prefix)
stopConsumer()
consumerCreatedTsc.TrySetException(ex) |> ignore
else
Log.Logger.LogInformation("{0} mailbox has stopped normally", prefix)
finally
// only after the Failed transition and cleanup, so a queued Close can't race with them
drainStoppedMailbox())
|> ignore

do
Expand All @@ -1046,7 +1133,10 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration
| _ -> ()


member private this.Mb with get(): Channel<MultiTopicConsumerMessage<'T>> = mb
member internal this.Mb with get(): Channel<MultiTopicConsumerMessage<'T>> = mb

member internal this.Consumers =
consumers.Values |> Seq.map fst |> Seq.toArray

member this.ConsumerId with get() = consumerId

Expand Down
79 changes: 71 additions & 8 deletions src/Pulsar.Client/Internal/PartitionedProducerImpl.fs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,10 +140,59 @@ type internal PartitionedProducerImpl<'T> private (producerConfig: ProducerConfi
Log.Logger.LogInformation("{0} stopped", prefix)

let mb = Channel.CreateUnbounded<PartitionedProducerMessage>(UnboundedChannelOptions(SingleReader = true, AllowSynchronousContinuations = true))

let replyFromStoppedMailbox (ex: exn) (msg: PartitionedProducerMessage) =
match msg with
| LastSequenceId channel ->
channel.TrySetException ex |> ignore
| GetStats channel ->
channel.TrySetException ex |> ignore
| LastDisconnectedTimestamp channel ->
channel.TrySetException ex |> ignore
| IsConnected channel ->
channel.TrySetResult false |> ignore
| Close channel ->
// the partition producers outlive a failed mailbox, so they still have to be closed.
backgroundTask {
try
let! _ =
producers
|> Seq.map (fun producer -> producer.DisposeAsync().AsTask())
|> Task.WhenAll
this.ConnectionState <- Closed
channel.TrySetResult(Ok ()) |> ignore
with Flatten closeEx ->
Log.Logger.LogError(closeEx, "{0} could not close all partition producers properly", prefix)
this.ConnectionState <- Closed
channel.TrySetResult(Error closeEx) |> ignore
} |> ignore
| Init | TickTime ->
()

let stoppedMailboxException () =
match this.ConnectionState with
| Closing | Closed -> AlreadyClosedException(prefix + " already closed") :> exn
| _ -> NotConnectedException(prefix + " not connected") :> exn

// requests posted after the mailbox exits would otherwise never be completed
let drainStoppedMailbox () =
backgroundTask {
while true do
let! msg = mb.Reader.ReadAsync()
Comment on lines +179 to +181
try
replyFromStoppedMailbox (stoppedMailboxException ()) msg
with ex ->
Log.Logger.LogWarning(ex, "{0} failed to reply to a request after the mailbox stopped", prefix)
} |> ignore

let mutable currentMessage : PartitionedProducerMessage option = None

do (backgroundTask {
let mutable continueLoop = true
while continueLoop do
match! mb.Reader.ReadAsync() with
let! msg = mb.Reader.ReadAsync()
currentMessage <- Some msg
match msg with
| Init ->
let producerTasks =
Seq.init numPartitions (fun partitionIndex ->
Expand Down Expand Up @@ -284,13 +333,22 @@ type internal PartitionedProducerImpl<'T> private (producerConfig: ProducerConfi
|> Task.WhenAll
channel.SetResult(statsReduce stats)
}:> Task).ContinueWith(fun t ->
if t.IsFaulted then
let (Flatten ex) = t.Exception
Log.Logger.LogCritical(ex, "{0} mailbox failure", prefix)
this.ConnectionState <- Failed
stopProducer()
else
Log.Logger.LogInformation("{0} mailbox has stopped normally", prefix))
try
if t.IsFaulted then
let (Flatten ex) = t.Exception
Log.Logger.LogCritical(ex, "{0} mailbox failure", prefix)
this.ConnectionState <- Failed
stopProducer()
try
currentMessage |> Option.iter (replyFromStoppedMailbox ex)
with currentEx ->
Log.Logger.LogWarning(currentEx, "{0} failed to fail the in-flight request after mailbox failure", prefix)
producerCreatedTsc.TrySetException(ex) |> ignore
else
Log.Logger.LogInformation("{0} mailbox has stopped normally", prefix)
finally
// only after the Failed transition and cleanup, so a queued Close can't race with them
drainStoppedMailbox())
|> ignore

do
Expand All @@ -305,6 +363,11 @@ type internal PartitionedProducerImpl<'T> private (producerConfig: ProducerConfi

override this.GetHashCode () = int producerId

member internal this.Producers =
producers.ToArray()

member internal this.Mb with get(): Channel<PartitionedProducerMessage> = mb

member private this.ChoosePartitionIfActive (message: MessageBuilder<'T>) =
match this.ConnectionState with
| Closing | Closed ->
Expand Down
Loading
Loading