Skip to content

Keep multi-topic consumers running when a child consumer stops - #384

Merged
Lanayx merged 6 commits into
developfrom
fix/multi-topic-child-failure-383
Oct 24, 2026
Merged

Lanayx merged 6 commits into
developfrom
fix/multi-topic-child-failure-383

Conversation

@Lanayx

@Lanayx Lanayx commented Oct 24, 2026 •

Copy link
Copy Markdown
Member

Fixes #383

Problem

Since 3.19.8, a multi-topic consumer moves to a permanent Failed state as soon as one child consumer stops. After that, every call throws mt/consumer(...) not connected, and the only way out is to recreate the consumer. 3.16.0 could not recover either: the poller died and was only logged, so the consumer silently stopped delivering. The difference is that 3.19.8 reports it. This PR fixes the underlying problem.

Two paths take a regex consumer down. Both start with a matched topic going away.

  1. A stopped child fails the parent's poller. Pattern children subscribe without creating topics, so when a matched topic is deleted, the child's resubscribe gets TopicDoesNotExistException and it moves to Failed. The next time the parent starts a read on that child (after a seek or redelivery, or when the poller picks up a read the child had already completed), the child throws NotConnectedException. That rethrow fails the poller and, through PollerFailed, the whole parent. A child mailbox failure took the same route.
  2. Acknowledging a message from a removed topic crashes the parent mailbox. Acknowledge, NegativeAcknowledge, AcknowledgeCumulative and ReconsumeLater* used consumers[topic] directly in the mailbox loop. After pattern discovery removed a topic, acking a message the application still held from it threw KeyNotFoundException and failed the mailbox.

Fix

  • A stopped child is detached instead of failing the parent. A child that returns, or throws, AlreadyClosedException/NotConnectedException posts ChildStopped, and its read parks. The parent removes it and then:
    • durable subscriptions, pattern consumers included: recreate the child with the client's backoff, off the mailbox, so the other topics keep being served. Until it is back, IsConnected and HasReachedEndOfTopic are false, and seeking or unsubscribing fails with NotConnectedException while the consumer keeps running. Skipping the child instead would report success while it kept its old position, or its subscription. Close and Unsubscribe wait for children that are still being created or disposed, so none stays subscribed after they return.
      Pattern children never create topics, so a deleted pattern topic can't be subscribed again and stays pending until the next discovery removes it. If the topic is recreated first, the next retry picks it up; before, a topic deleted and recreated between two discovery runs was never consumed again.
    • non-durable pattern consumers (non-partitioned child): drop the topic from the known topics, so the next discovery subscribes again if it still matches. The re-subscribed topic starts from the initial position, like any newly discovered topic.
    • other non-durable subscriptions (e.g. multi-topic readers): fail the parent, because a recreated child would restart from the initial position.
    • after a failed Unsubscribe, which leaves the consumer running so the call can be retried: fail the parent, because the child may already have been unsubscribed and must not be recreated. Ignoring the stopped child would leave its read, and pending receives, waiting forever. Before this PR, the poller failure already failed the parent in this case.
  • New and recreated children start at the parent's consumer epoch. For Exclusive/Failover subscriptions the parent advances its epoch on every seek and redelivery, and it discards messages with an older epoch. A child created afterwards subscribed with epoch 0, so the parent silently dropped everything from it. This was already the case for partitions added by AutoUpdatePartitions and topics added by pattern discovery; recreation would have made it common.
  • Children no longer dispose the parent's interceptors. Each child disposed the shared interceptors when it closed. Once one child stopped, or pattern discovery removed a topic, the remaining children kept calling disposed interceptors; OtelConsumerInterceptor, for example, stops tracing for good once disposed. Children now get a non-owning view, and the parent disposes the interceptors once, when it stops.
  • Ack, nack, cumulative ack and reconsume for a topic without a consumer fail only that request, with NotConnectedException. The consumer keeps running. As with a single-topic consumer, a failed ack, cumulative ack or reconsume is passed to the interceptors' OnAcknowledge/OnAcknowledgeCumulative, so, for example, tracing ends for that message.
  • Terminal failures keep their cause. NotConnectedException and AlreadyClosedException get an (message, innerException) constructor. A failed multi-topic consumer reports ... has failed and can't recover, it has to be recreated: <cause> and keeps the original exception as InnerException. ConnectionHandler remembers why it failed, so a failed child's errors carry the cause up to the parent too.
  • A consumer whose topic no longer exists fails its pending receives with NotConnectedException, instead of leaving them waiting forever. This is how the parent learns a child has stopped. It also changes single-topic consumers: a pending ReceiveAsync on a deleted topic, when topic auto-creation is disabled, now fails instead of hanging.

Related fixes in code this PR touches:

  • Child registration happens on the mailbox. Before, consumers.Add ran lazily inside sequences that TaskSeq's own thread consumed, racing the mailbox's reads of the same Dictionary after discovery or a partition update.
  • handlePartitions enumerated its consumer-creation sequence a second time on failure, after the partition count had already been raised. Partly created consumers were never disposed, and the failed partitions were never retried. Consumers are now created once, and the new count is recorded only after all of them succeed. The receiver queue split now counts the new partitions, so it can't divide by zero when no child is currently registered. A failing partition check is logged instead of crashing the mailbox, as pattern discovery already did.
  • processRemovedTopics matches children by their parent topic. It used to strip -partition- from child names, which did not match pattern topics that are themselves partition names. It also always forgets removed topics, even when they have no live child.
  • LastDisconnectedTimestamp returns 0 when there are no children (e.g. a pattern that matches nothing) instead of failing the mailbox.
  • ReconsumeLaterCumulative never completed its reply on success.

Tests

Integration tests that expected the old behaviour (a child failure stops the parent; an ack for an unknown topic fails the mailbox) were rewritten:

  • Multi-topic consumer recreates a child whose mailbox fails: after a redelivery has moved the consumer epoch, a pending receive survives the failure, then gets a message from the recreated child.
  • Child mailbox failure keeps a pending batch receive alive: the pending batch completes from the recreated child.
  • Multi-topic consumer closes a recreated child at the broker
  • Non-durable multi-topic consumer stops with the cause when a child mailbox fails
  • Multi-topic consumer completes the request that faults its mailbox and Close queued behind a multi-topic consumer mailbox failure completes only after cleanup now fault the mailbox with a null poll result instead of a bogus ack. The first also checks that the cause is kept.

New tests:

  • Multi-topic consumer fails only the acknowledgement for an unknown topic, which also checks that the interceptors are told about the failed acks
  • Acknowledging a message from a removed pattern topic fails only that request
  • Pattern consumer keeps running after a matched topic is deleted: force-deletes a matched topic, restarts every stream with a redelivery (this used to fail the consumer), checks that seeking and unsubscribing fail while the topic is pending, keeps consuming the other topic, then recreates the topic and checks it is subscribed again without waiting for discovery.
  • Multi-topic consumer stops when a child stops after a failed unsubscribe: one child unsubscribes and the other has failed, so Unsubscribe fails; the pending receive then fails with AlreadyClosedException instead of waiting forever.
  • Unsubscribing waits for a child that is still being recreated for a removed topic: holds the recreated child's broker lookup while discovery removes its topic, and checks Unsubscribe doesn't complete until that child is created and closed.
  • A child recreated across a redelivery is created again at the new epoch: holds the recreated child's broker lookup while a redelivery moves the epoch on, checks the stale child is discarded and created again, that the new child's messages are delivered, and that neither child is still subscribed after Close.
  • Closing a multi-topic consumer waits for a child that is being recreated: holds the recreated child's broker lookup, checks Close doesn't complete until that child is created and closed, then subscribes a new exclusive consumer to the topic.
  • Multi-topic consumer refuses to seek or unsubscribe while a child is being recreated: uses a long client backoff so the failed child stays pending, and checks the consumer keeps consuming the other topic.
  • Recreating a multi-topic consumer's child keeps the interceptors: the recreated child still runs the interceptors, and they are disposed exactly once, when the parent closes.

unwrap and assertNotConnected moved from Basic.fs to Common.fs so MultiTopic.fs can use them, next to the new assertNotConnectedBecauseOf and waitUntil helpers.

Verification

  • dotnet build -c Release succeeds with no new warnings.
  • Unit tests: 236 passed, 2 ignored.
  • I did not run the integration tests locally; they run in this PR's Ubuntu CI.

🤖 Generated with Claude Code

A stopped child consumer used to fail the parent's poller, and acknowledging
a message from a removed topic crashed the parent mailbox. Either left the
multi-topic consumer permanently failed with "not connected" (#383).

- Detach a stopped child instead of failing the parent. Pattern consumers
  re-subscribe on the next discovery, other durable consumers recreate the
  child with backoff, and non-durable ones still fail, now with the cause.
- Start new and recreated children at the parent's consumer epoch, so their
  messages aren't discarded as stale after a seek or redelivery.
- Fail only the request when acknowledging, negatively acknowledging or
  reconsuming a message for a topic without a consumer.
- Keep the cause of terminal failures as the inner exception.
- Fail pending receives when a consumer's topic no longer exists.
- Register children on the mailbox, create partition consumers once, match
  removed pattern topics by their topic name, handle zero children in
  LastDisconnectedTimestamp, and complete ReconsumeLaterCumulative.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Changes recommended

Recovery still has unresolved shutdown, interceptor lifetime, unsubscribe, and end-of-topic correctness issues.

4 open issues
What changed in this PR

Addresses #383 by allowing multi-topic consumers to recover from stopped children while healthy topics continue delivering messages.

Changes:

  • Adds child detachment and recreation, preserving consumer epochs across recovery.
  • Isolates requests for removed topics and preserves terminal failure causes.
  • Expands integration coverage for recovery, topic deletion, acknowledgements, and cleanup.
File Description
tests/​IntegrationTests/​MultiTopic.fs Tests pattern-topic removal, deletion, and rediscovery.
tests/​IntegrationTests/​Common.fs Shares failure assertions and polling helpers.
tests/​IntegrationTests/​Basic.fs Tests child recovery, terminal causes, and broker cleanup.
src/​Pulsar.Client/​Internal/​MultiTopicsConsumerImpl.fs Implements child recovery and safer request handling.
src/​Pulsar.Client/​Internal/​ConsumerImpl.fs Propagates failure causes and initializes child epochs.
src/​Pulsar.Client/​Internal/​ConnectionHandler.fs Retains and exposes connection failure causes.
src/​Pulsar.Client/​Api/​PulsarClientException.fs Adds constructors accepting inner exceptions.

Review effort: Balanced

…rceptors

- Close waits for children that are still being recreated and closes them,
  so none stays subscribed after Close returns.
- Unsubscribe fails without stopping the consumer while a child is being
  recreated, instead of leaving that child's subscription behind.
- HasReachedEndOfTopic is false while a child is being recreated.
- Children get a non-owning view of the interceptors, and the parent
  disposes them once when it stops. A stopped child used to dispose the
  interceptors the other children were still using.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Changes recommended

Shutdown can miss outstanding child disposals, and stopped pattern children can be silently skipped by seek and unsubscribe operations.

1 open issue
1 previously missed issue

In code that hasn't changed since last review

Medium severity Track failed durable child topics as pending

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

When a durable pattern child's mailbox fails while its topic still exists, this branch forgets it without adding it to pendingChildren. Before the next discovery, SeekAsync and UnsubscribeAsync therefore pass their pending-child guards and report success after operating only on surviving topics, leaving the failed topic's cursor or durable subscription unchanged. Retain a pending marker until discovery confirms removal or successfully re-subscribes the topic, and include it in these guards.

4 resolved since last review

Review effort: Balanced

…en pending

- Children disposed without blocking the mailbox, e.g. a recreated child
  discarded for a stale epoch, are tracked, and Close waits for them.
- A stopped child of a durable pattern consumer stays pending and is
  recreated like any other child, instead of being forgotten until the next
  discovery. Seek and Unsubscribe no longer skip it, and a topic recreated
  before discovery runs is picked up by the retry. A deleted topic stays
  pending until discovery removes it. Non-durable pattern consumers still
  leave a stopped topic to discovery.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Changes recommended

Unsubscribe can complete before child cleanup finishes, and stalled polling conditions can hang the recovery tests.

2 open issues
1 resolved since last review

Review effort: Balanced

Comment thread tests/IntegrationTests/Common.fs Resolved
- Unsubscribe also waits for children still being recreated or disposed,
  e.g. a child of a topic that discovery removed while it was being
  recreated, so none stays subscribed after it returns.
- waitUntil bounds each condition by the remaining deadline, so a mailbox
  that stops replying fails the test instead of hanging it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 Needs a closer look

An unresolved lifecycle race and intertwined recovery and shutdown paths require human review and broker-backed integration verification.

No open issues 🎉

1 previously missed issue

In code that hasn't changed since last review

Medium severity Handle failed unsubscribe to cancel parked reads and receive waiters

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

If Unsubscribe fails before a queued ChildStopped is handled, it leaves the mailbox running in Failed (lines 1260–1262). This branch then ignores the stopped child without releasing its parked read or cancelling the receive token, so pending receives can hang indefinitely. Handle Failed explicitly by calling failConsumer ex and exiting the mailbox loop, which cancels parked reads and fails receive waiters.

2 resolved since last review

Review effort: Balanced

…nsubscribe

A failed Unsubscribe leaves the mailbox running in the Failed state so the
call can be retried. A child that stopped in that state was ignored, leaving
its parked read and the pending receives waiting forever. It may already
have been unsubscribed, so it can't be recreated: the consumer now stops,
which fails the pending receives. A stopped child that is no longer
registered has its parked read released in any state.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 Needs a closer look

Missing-child request failures omit interceptor notifications, potentially leaving tracing activities open for the lifetime of the parent consumer.

No open issues 🎉

1 previously missed issue

In code that hasn't changed since last review

Medium severity Notify interceptors when acknowledgement requests fail

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

Missing-child acknowledgement and reconsume failures bypass the interceptor callbacks, so a retained message from a removed topic can leave its OpenTelemetry activity cached until parent shutdown when acknowledgement timeouts are disabled. Before failing each request, notify OnAcknowledge for individual operations or OnAcknowledgeCumulative for cumulative operations with the same exception. The same issue occurs in this file at lines 994, 1187, and 1203.

Review effort: Balanced

An ack, cumulative ack or reconsume for a topic without a consumer now
calls OnAcknowledge or OnAcknowledgeCumulative with the exception, as a
single-topic consumer does, so that e.g. the tracing activity for the
message ends instead of staying open until the consumer is disposed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 Needs a closer look

Concurrent child recovery, discovery, epoch changes, and shutdown coordination need final human review supported by confirmed broker-backed integration results.

No open issues 🎉

Review effort: Balanced

@Lanayx
Lanayx merged commit 423ef4f into develop Oct 24, 2026
5 checks passed
Lanayx added a commit that referenced this pull request Oct 24, 2026
A multi-topic consumer no longer recreates a child that has stopped,
which partly reverts #384. As in the Java client, the parent stops
reading a closed child and retries any other receive failure after 10
seconds, so a child never fails the parent. The stopped child stays
registered until the parent closes or discovery removes its topic, so
IsConnected is false and Seek and Unsubscribe fail in the meantime.
Redelivery skips it, and acknowledging a message from a removed topic
no longer notifies the interceptors.

A producer or consumer whose topic was deleted, and which may not
create it again, now closes itself instead of reconnecting. Pending
sends and receives fail with AlreadyClosedException.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Lanayx added a commit that referenced this pull request Oct 25, 2026
)

* Keep the cause when a producer fails

A producer whose mailbox failed answered every later send, flush and
stats request with a bare "not connected", so the reason it failed was
only in the log. ProducerImpl now passes the cause to its connection
handler, which already reports it on later requests, and the stopped
mailbox includes it too. A partitioned producer whose mailbox failed
reports "has failed and can't recover, it has to be recreated: <cause>"
with the original exception as InnerException, like the multi-topic
consumer does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Handle stopped children and deleted topics as the Java client does

A multi-topic consumer no longer recreates a child that has stopped,
which partly reverts #384. As in the Java client, the parent stops
reading a closed child and retries any other receive failure after 10
seconds, so a child never fails the parent. The stopped child stays
registered until the parent closes or discovery removes its topic, so
IsConnected is false and Seek and Unsubscribe fail in the meantime.
Redelivery skips it, and acknowledging a message from a removed topic
no longer notifies the interceptors.

A producer or consumer whose topic was deleted, and which may not
create it again, now closes itself instead of reconnecting. Pending
sends and receives fail with AlreadyClosedException.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Address review: shared producer interceptors, partition stats, removed children

- Partition producers get a non-owning view of the interceptors, and the
  partitioned producer disposes them once, as in the Java client. A
  partition that closed itself used to dispose the interceptors the
  other partitions were still using.
- A stopped partition producer fails only the partitioned producer's
  GetStats request instead of its mailbox, so the other partitions keep
  working.
- Removing a multi-topic consumer's child ends the read that was waiting
  on it, or retrying it, instead of leaving it until the next reset.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Address review: HTTP lookup 404s, interceptors disposed last

- The HTTP lookup reports a 404 as NotFoundException, as the Java client
  does, so a producer or consumer whose topic was deleted stops instead
  of reconnecting when the client uses an HTTP service URL. The schema
  lookup still reads a 404 as "no schema".
- A producer fails its pending messages before it disposes the
  interceptors, and disposes them last, as in the Java client, so their
  OnSendAcknowledgement still runs for those messages. This applies to
  Close and to the producer closing itself because its topic was
  deleted.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Address review: dispose interceptors after the children are closed

After a mailbox failure, a partitioned producer or multi-topic consumer
keeps its children running until Close, so the interceptors they borrow
are now disposed only once the children are closed. Interceptor Close is
idempotent, as every Close of a stopped owner reaches it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Address review: overlapping closes and failed partition closes

After a mailbox failure, the partitioned producer and multi-topic
consumer now handle Close as the running mailbox and the Java client do:
the first Close moves to Closing, closes the children and then disposes
the interceptors, and a Close while closing returns at once instead of
running a second cleanup that could dispose the interceptors early.

As in the Java client, a partitioned producer whose partition fails to
close still disposes the interceptors once every partition has finished.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Address review: registration of failed parents, failed sends, test race

A partitioned producer or multi-topic consumer whose mailbox or poller
failed now stays registered with the client until it is closed, as in
the Java client, so closing only the client still closes its children
and disposes its interceptors.

As in the Java client, failing the pending sends now also fails batched
and blocked sends through OnSendAcknowledgement.

The pattern removal test waits for discovery to finish before it checks
the remaining children.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Move the batched-send test next to the other close tests

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
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.

Multi-topic/regex consumer remains unusable after BatchReceiveAsync NotConnectedException in 3.19.8/3.19.9

2 participants