Pulsar: fix KeyNotFoundException acking batch messages on partitioned topics (supersedes #2883)

[jeremydmiller]

Supersedes #2883. Co-authored with @patrick-cloke-simplisafe, who diagnosed the upstream DotPulsar bug and wrote the original fix.

The bug

DotPulsar's BatchHandler.Add (apache/pulsar-dotpulsar#287) constructs the inner MessageId for each message of a batch via the four-arg ctor and never sets MessageId.Topic. Consumer.Acknowledge then does _subConsumers[messageId.Topic] with the empty string and throws KeyNotFoundException on partitioned topics, so batched messages on partitioned topics are never acked → infinite redelivery.

Non-batched messages are unaffected: ConsumerChannel uses ToMessageId(topic) for those, which correctly populates Topic.

Fix

PulsarListener.FixMessageId reconstructs the MessageId with the partition sub-topic URI ({topic}-partition-{N}) when Topic is empty and Partition >= 0. Short-circuits and returns the original MessageId instance otherwise (forward-compat for when the upstream DotPulsar fix lands; non-partitioned messages pass through unchanged). Applied at the three Acknowledge call sites in PulsarListener: CompleteAsync, DeferAsync, and moveToQueueAsync (the DLQ / retry-letter path).

Difference from #2883

One substantive change: each ack call site uses the source consumer's own IConsumer.Topic (consumer.Topic / sourceConsumer.Topic) rather than _endpoint.PulsarTopic(). DotPulsar exposes Topic on the IConsumer abstraction (verified in DotPulsar/Abstractions/IConsumer.cs; stable since 3.x), so the same helper works correctly for the retry consumer without any branching. The retry consumer is subscribed to {baseTopic}-RETRY (or a user-configured retry topic), so the original _endpoint.PulsarTopic()-based reconstruction would have produced wrong sub-topic names ({baseTopic}-partition-N instead of {baseTopic}-RETRY-partition-N) and re-triggered the same KeyNotFoundException on the retry path for users with partitioned retry topics. Latent in the original PR; addressed here.

Carries Patrick's three unit tests verbatim, plus one new test (FixMessageId_BatchMessageOnPartitionedRetryTopic_UsesRetryTopicForReconstruction) that pins the retry-topic reconstruction contract.

On Envelope.TopicName

@jeremydmiller asked on #2883 whether Envelope.TopicName could carry the topic and avoid URI reconstruction entirely. It can't. DotPulsar's IMessage<T> has no Topic property — the concrete Message<T> ctor takes no topic parameter, and MessageFactory<T> doesn't even know the topic it's constructing for. The only sources for the partition sub-topic at the ack site are (a) MessageId.Topic when set (works for non-batched) and (b) reconstruction from the consumer's base topic + MessageId.Partition. Populating Envelope.TopicName from that same reconstruction would just centralize the formatting in PulsarEnvelopeMapper; it doesn't eliminate it.

A separate enhancement to surface topic info to user handlers via Envelope.TopicName on incoming Pulsar messages is worth pursuing in its own PR — but it's out of scope for the bug fix here.

Verification

• FixMessageIdTests: 4/4 pass locally (net9.0).
• Full dotnet build wolverine.slnx -c Release clean (0 warnings, 0 errors).
• The wider Pulsar compliance suite (PulsarTransportComplianceTests, InlinePulsarTransportComplianceTests, with_cloud_events) has 11 pre-existing local failures around retry/DLQ — verified that the same tests fail identically on baseline origin/main with my fix stashed (BadImageFormatException during handler load → Wolverine runtime-codegen issue with Apple-Silicon macOS, unrelated to this change). CI on Linux x64 validates them.

🤖 Generated with Claude Code