Update RocketMQ.Client to 5.2.1 and fix RocketMQ test suite

[lillo42]

Description

Updates the RocketMQ client (RocketMQ.Client 5.1.0 → 5.2.1) and fixes the RocketMQ test suite, which had been silently broken (the RocketMQ CI job is currently disabled, so regressions went unnoticed).

Client upgrade (772490a7a):

• Bump RocketMQ.Client to 5.2.1 and migrate RocketMessageConsumer to its APIs — real Nack via ChangeInvisibleDuration instead of a no-op, and primary-constructor parameter capture in place of explicit fields.

Test suite fixes (4990be010):

• Producer: empty Baggage property rejected. RocketMqMessageProducer unconditionally added the CE_baggage property; an empty Baggage (the default) stringifies to "", which the client's Message.Builder.AddProperty rejects (value should not be null or white space, validation present since client 5.1.0). Every Send/SendAsync threw — this alone accounted for 24 of 35 test failures. The property is now only added when the baggage has entries.
• Topic map typos. RocketMqMessageGatewayProvider.s_topicMap used When_requeing_... keys vs. the actual When_requeuing_... test method names, so four requeue tests fell through to non-existent topics and died with NotFoundException.
• RocketConsumerFactoryDlqTests rework. The test could never pass: it built a subscription without a consumer group (which defaulted to string.Empty and fails the client's SetConsumerGroup regex), and it asserted via reflection on private fields removed in the client-upgrade commit. It now passes a random consumer group and asserts on the new public RocketMessageConsumer.DeadLetterRoutingKey / InvalidMessageRoutingKey properties instead of implementation details.
• docker-compose-rocketmq.yaml: image pinned (apache/rocketmq:5.4.0, matching the hardcoded mqadmin paths), proxy heap raised to -Xmx512m, orders/orders-dlq/orders-invalid topics added, and misleading port comments corrected (8081 is the gRPC endpoint clients use).

Related Issues

Type of Change

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)
• Documentation update

Checklist

• I have read the Contributing Guide
• I have checked the documentation for relevant guidance
• I have added/updated XML documentation for any public API changes
• I have added/updated tests as appropriate
• My changes follow the existing code style and conventions

Additional Notes

• Verified against a local RocketMQ broker (topics provisioned by the compose file with correct message.type attributes): all code-level failures are resolved, including the previously broken DLQ/reject and consumer-factory tests.
• Remaining suite instability is environmental, not Brighter code: ReceiveMessage long-polls through the local podman-forwarded proxy intermittently hang until the client deadline (HTTP/2 CANCEL → DeadlineExceeded). Reproduced standalone with the raw 5.2.1 client (no Brighter code) on both broker 5.4.0 and 5.5.0 — likely the rootless port-forwarding breaking long-lived gRPC streams, or a proxy-side long-poll issue (cf. [Bug] Missing long-polling notification under CombineConsumeQueue selective double-write mode apache/rocketmq#10615, fixed in the upcoming 5.5.1). Tests that receive timely broker responses all pass.

[claude]

Review — Update rocketmq

Nice, focused PR. Three of the changes are unambiguous wins:

• Pinning apache/rocketmq:5.4.0 fixes a latent break — the create-topic step hardcodes /home/rocketmq/rocketmq-5.4.0/bin/mqadmin, so the compose file would have silently broken the moment latest moved past 5.4.0.
• The requeing → requeuing key fix in RocketMqMessageGatewayProvider is a real bug fix. Those keys never matched the generated test names (tests/.../Generated/Reactor/When_requeuing_a_failed_message_should_receive_message_again.cs), so GetOrCreateRoutingKey() was falling through to the gen_nonexistent_{guid} branch and the four requeue tests were quietly running against throwaway auto-created topics instead of gen_r_requeue / gen_p_requeue.
• Replacing reflection with public properties in the DLQ factory test — much better than BindingFlags.NonPublic, and the Baggage.Any() guard matches the existing convention in RmqMessagePublisher.cs:201 and KafkaDefaultMessageHeaderBuilder.cs:106.

A few things I'd want resolved before merge.

---

1. Requeue is still a no-op that reports success, but Nack now uses the API it says is broken

RocketMessageConsumer.cs:196-201:

// Waiting for next RocketMQ C# version, due an issue on ChangeInvisibleDuration
// consumer.ChangeInvisibleDuration(view, TimeSpan.Zero);
return true;

…while NackAsync (:116-124) now calls exactly that:

await consumer.ChangeInvisibleDuration(view, TimeSpan.Zero);

This PR is the "next RocketMQ C# version" (5.1.0 → 5.2.1). Both can't be right: either the client bug is fixed and Requeue should be implemented, or it isn't and Nack is broken. Right now Requeue returns true — telling the pump the requeue succeeded — having done nothing, and it also ignores the delay argument entirely.

This isn't theoretical, and it interacts with the topic-map fix above. When_requeuing_a_failed_message_should_receive_message_again now resolves to the real gen_r_requeue topic, calls _channel.Requeue(received), sleeps 5s, then polls 10 × 300ms. But RocketMqSubscription.cs:105 defaults InvisibilityTimeout to 30 seconds, so with Requeue as a no-op the message cannot come back inside the ~8s the test allows. Same for the _with_delay and _async variants. Implementing Requeue via ChangeInvisibleDuration(view, delay ?? TimeSpan.Zero) would make the fixed mapping actually pay off; RequeueAsync should then be async-first rather than Task.FromResult(Requeue(...)).

If ChangeInvisibleDuration is still broken for the requeue case specifically, please update the stale comment to say so and explain why Nack is safe.

2. NackAsync has no error handling or logging, and it's called from inside a catch block

Compare SqsMessageConsumer.NackAsync (src/Paramore.Brighter.MessagingGateway.AWSSQS/SqsMessageConsumer.cs:334-360), which has the same "set visibility to zero" semantics but wraps the call, specifically handles the stale-receipt-handle case, and logs on both entry and success.

The RocketMQ version does none of that. That matters here because both pumps invoke Nack from inside an exception handler — Reactor.cs:328 and Proactor.cs:362 are in catch (DontAckAction) blocks. An exception thrown there escapes the handler rather than being caught by it. Previously Nack was a guaranteed-no-throw no-op, so this is a genuinely new failure mode: a transient gRPC/broker error during nack can now tear down the message pump instead of being retried on the next receive.

Suggest mirroring the SQS shape — try/catch, log, swallow. The class already has the Log partial to hang NackingMessage/NackedMessage/ErrorNackingMessage off, and there's currently no logging on this path at all. Worth a short comment on why cancellationToken is unused too (the client API doesn't take one), since the parameter is silently dropped.

3. Test coverage

CLAUDE.md is explicit that behaviour changes go through /test-first. The Nack change is a behaviour change from "no-op" to "immediate redelivery", and there is no Nack test anywhere in Paramore.Brighter.RocketMQ.Tests — nothing covers the happy path, the missing-ReceiptHandle early return, or the new throw-on-broker-error path. The empty-Baggage guard is likewise untested.

Also worth flagging for reviewers: the RocketMQ suite is commented out in CI (ci.yml:687, "TODO: Rafael Andrade is working on how to run RocketMQ on GHA"), so none of this is verified by the pipeline. Could you note in the PR description how you validated it locally — particularly whether the requeue tests now pass?

4. The test lost its connection assertion with no replacement

The old reflection-based test asserted three things; the new one asserts two. The dropped one was Assert.NotNull(actualConnection).

That assertion was load-bearing. CreateProducerAsync returns null when connection == null (RocketMessageConsumer.cs:236-237), which makes DLQ/invalid routing silently degrade to "log a warning and ack the source message" — a regression in RocketMessageConsumerFactory that dropped the connection would now pass this test green. Either expose the connection alongside the routing keys, or replace it with a behavioural assertion.

Minor, related: the test now needs a live broker just to assert constructor passthrough (_factory.Create builds a real SimpleConsumer), which is why the orders* topics had to be added to compose. It also creates a fresh Guid consumer group per run, which accumulates subscription groups on the broker. Not blocking, but a consumer that could be constructed without a broker would make this a real unit test.

5. Nearby risk the Baggage guard hints at

I'm assuming the guard was added because 5.2.1 rejects empty property values. If so, RocketMqMessageProducer.cs:152-158 is the next landmine — it forwards every non-local bag entry via val.ToString(), and the consumer stuffs a live MessageView object into the bag at RocketMessageConsumer.cs:338:

header.Bag["ReceiptHandle"] = message;

MessageHeader.IsLocalHeader won't filter it — nothing in the codebase calls RegisterLocalHeader at all. So on the reject/DLQ path (which this PR makes considerably more live) the republished message carries a ReceiptHandle property containing MessageView.ToString(), and a null bag value would NRE. MessageHeader.RegisterLocalHeader("ReceiptHandle") at gateway init looks like the intended mechanism. Pre-existing, so fine as a follow-up.

While you're in that file: HeaderNames.Type is added twice (:94 and :109) whenever Header.Type is non-empty — also pre-existing, but it's four lines from your change.

6. Minor

• BOM removed from RocketMessageConsumer.cs and RocketMqMessageGatewayProvider.cs. .editorconfig:15 sets charset = utf-8-bom. Unrelated diff noise that also breaks the stated convention — worth restoring.
• Switch-expression formatting (:277-283): trailing whitespace after switch, a double space in DeadLetterRoutingKey != null (:280), and missing spaces in => (null, ...) (:281) and =>(DeadLetterRoutingKey, ...) (:282). The rewrite itself is behaviour-preserving and reads better than the old switch statement.
• Tuple naming: DetermineRejectionRoute declares bool foundProducer (:274) but every call site destructures it as shouldRoute (:156) — shouldRoute is the accurate name, and you're already touching this method.
• Nack uses a block body while Acknowledge, Purge, and Receive right above it all use expression bodies.
• Port 9877 is now annotated # Reserved (no service listens here by default). If nothing listens, drop the mapping rather than document a dead port. (The 8081 correction to "gRPC endpoint" is right and matches GatewayFactory.cs:11's SetEndpoints("localhost:8081") — the old comments had the two ports swapped.)
• Proxy -Xmx512m — a 4× bump while MaxDirectMemorySize=64m / MaxMetaspaceSize=96m stay put and the broker stays at 128m. Was this an OOM under 5.4.0? A one-line comment would save the next person the archaeology.

---

Happy to re-review once (1) and (2) are settled — those are the two that change runtime behaviour.