fix(broker): give repeat same-topic subscriptions distinct consumer groups (EVO-1737)

[nickoliveira23]

Summary

npm run dev:single crashed on boot with KafkaBrokerAdapter already has a consumer registered for topic "campaigns.control". In single mode every runner loads in one process, so the CampaignsControlConsumer registered in both campaign-packer.module.ts and campaign-sender.module.ts (EVO-1222 — correct by design for the distributed deployment) collided on one consumer group.

Fix: KafkaBrokerAdapter.subscribe now gives each repeat subscription to the same topic its own consumer group (<group>, then <group>-2, …) so both consumers coexist and each receives every message — the same broadcast the distributed deployment already gets from separate per-runner groups. The dispatchMessage/ack hot path is untouched; the per-topic counter is cleared on shutdown.

Changes

• kafka-broker.adapter.ts: per-topic subscription counter → distinct group for repeat subscriptions; map keyed by group; counter cleared in onModuleDestroy.
• kafka-broker.adapter.spec.ts: replaced the old "rejects subscribing twice" test with one asserting two distinct groups and that both consumers receive independently.

Validation

• evo-flow: npm run typecheck → clean (exit 0)
• evo-flow: npx jest src/shared/broker/adapters/kafka-broker.adapter.spec.ts → 23/23 pass (incl. new EVO-1737 test)
• evo-flow: npm run dev:single → reaches 🎯 Service ready in SINGLE mode; both campaigns.control consumers join distinct groups (single-campaigns.control + single-campaigns.control-2); 0 double-registration crashes.

Acceptance Criteria

• dev:single boots with both packer and sender active — no crash.
• Distributed mode unchanged (separate processes → distinct base groups, no suffix).
• pause/stop/resume control honored in single mode (both control consumers receive every message).

Known limitations (out of scope — noted from code review)

• M2: getTopicLag(topic) measures only the base group ${runMode}-${topic}; lag of a repeat-subscription group (…-2) is not surfaced. Only affects campaigns.control today (low-volume control fast-path).
• L2: subscribePattern still rejects a duplicate prefix — the same single-mode collision class would apply if two runners ever subscribed the same wildcard prefix (no current case).
• The -N suffix is positional (module load order), so group names are not guaranteed stable across restarts; acceptable because repeat-subscribed topics are broadcast control topics with an authoritative Postgres flag.

Linked Issue

• EVO-1737

🤖 Generated with Claude Code

Summary by Sourcery

Allow multiple consumers in a single process to subscribe to the same Kafka topic by assigning distinct consumer groups to repeat subscriptions.

Bug Fixes:

• Prevent crashes in single-process mode caused by duplicate Kafka consumers sharing the same group for a topic.
• Ensure both control consumers on the same topic can coexist and each receive all messages in single-runner deployments.

Enhancements:

• Track per-topic subscription counts to derive unique consumer group IDs for repeat subscriptions and clear this state on shutdown.

Tests:

• Replace the duplicate-subscription rejection test with a new test verifying distinct consumer groups and independent message delivery for repeat subscriptions.

[sourcery-ai]

Reviewer's Guide

KafkaBrokerAdapter now supports multiple subscriptions to the same topic within a single process by assigning unique consumer group IDs per subscription, and tests have been updated to assert the new behavior.

Sequence diagram for repeat topic subscription group assignment

sequenceDiagram
  actor Module
  participant KafkaBrokerAdapter
  participant topicSubscriptions
  participant consumers

  Module->>KafkaBrokerAdapter: subscribe("events-topic", handler1)
  KafkaBrokerAdapter->>KafkaBrokerAdapter: ensureTopicExists("events-topic")
  KafkaBrokerAdapter->>topicSubscriptions: get("events-topic") = 0
  KafkaBrokerAdapter->>KafkaBrokerAdapter: baseGroupId = resolveRunMode("events-topic") + "-events-topic"
  KafkaBrokerAdapter->>KafkaBrokerAdapter: groupId = baseGroupId
  KafkaBrokerAdapter->>consumers: has(groupId) = false
  KafkaBrokerAdapter->>topicSubscriptions: set("events-topic", 1)
  KafkaBrokerAdapter->>consumers: set(groupId, consumer1)

  Module->>KafkaBrokerAdapter: subscribe("events-topic", handler2)
  KafkaBrokerAdapter->>KafkaBrokerAdapter: ensureTopicExists("events-topic")
  KafkaBrokerAdapter->>topicSubscriptions: get("events-topic") = 1
  KafkaBrokerAdapter->>KafkaBrokerAdapter: baseGroupId = resolveRunMode("events-topic") + "-events-topic"
  KafkaBrokerAdapter->>KafkaBrokerAdapter: groupId = baseGroupId + "-2"
  KafkaBrokerAdapter->>consumers: has(groupId) = false
  KafkaBrokerAdapter->>topicSubscriptions: set("events-topic", 2)
  KafkaBrokerAdapter->>consumers: set(groupId, consumer2)

Loading

File-Level Changes

Change	Details	Files
Allow repeated subscriptions to the same Kafka topic by assigning distinct consumer groups per subscription and tracking them per topic.	Introduce a topicSubscriptions map to track subscription counts per topic. Change consumer map to be keyed by consumer group ID instead of topic. Compute a base consumer group ID from run mode and topic, and append a numeric suffix for repeat subscriptions. Ensure repeated subscriptions throw only when a consumer with the computed group ID already exists. Clear topicSubscriptions during module shutdown and log consumer shutdowns using the consumer key.	src/shared/broker/adapters/kafka-broker.adapter.ts
Update Kafka broker adapter tests to validate multiple-consumer behavior and consumer group naming for repeat subscriptions.	Replace the previous test that rejected duplicate topic subscriptions with one that allows a second subscription and checks for distinct group IDs. Assert that both consumers receive messages independently when subscribed to the same topic. Use the underlying Kafka test harness to inspect created consumer group IDs and consumers.	src/shared/broker/adapters/kafka-broker.adapter.spec.ts

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[sourcery-ai]

Hey - I've left some high level feedback:

• Consider including the topic (and perhaps run mode) in the duplicate-consumer error message instead of only the groupId, as debugging collisions will usually start from the topic name rather than the derived group string.
• The topicSubscriptions counter is only ever incremented and cleared on module destroy; if you later introduce unsubscribe-style behavior or partial subscription failures, it may be worth aligning the counter with actual active consumers to avoid unbounded suffix growth or mismatches.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- Consider including the topic (and perhaps run mode) in the duplicate-consumer error message instead of only the groupId, as debugging collisions will usually start from the topic name rather than the derived group string.
- The `topicSubscriptions` counter is only ever incremented and cleared on module destroy; if you later introduce unsubscribe-style behavior or partial subscription failures, it may be worth aligning the counter with actual active consumers to avoid unbounded suffix growth or mismatches.

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[dpaes]

✅ Approved — EVO-1737

Reviewed against the 3 ACs (dev:single boots both packer+sender with no campaigns.control double-registration; distributed unchanged; pause/stop/resume honored in single mode).

Core correctness — the crux is right: the fix gives each repeat same-topic subscription its own consumer group (<base>, then <base>-2) — group-per-handler, not a second consumer in the same group. Distinct Kafka groups each receive all messages (broadcast), so both packer and sender control consumers get every pause/stop/resume in single mode. This matches the fan-out distributed mode already gets from separate per-runner groups.

Verified:

• First subscription untouched — base group keeps its original name (only repeats get the -N suffix); no committed-offset reset on deploy. Asserted by the new + existing tests.
• Distributed mode unchanged (AC#2) — the per-topic counter is per-adapter-instance in-process state; separate processes never collide and never get a suffix.
• Dispatch/ack hot path untouched — only the map key changed (topic → groupId).
• Test proves fan-out — the new test asserts two distinct groups and that both consumers independently receive a message (not merely that group names differ).
• Disclosed limitations (getTopicLag base-group-only; subscribePattern dup-prefix) confirmed non-biting today (only campaigns.control hits this path).

Non-blocking follow-up: RabbitMQBrokerAdapter.subscribe still throws on a duplicate topic, so BROKER_TYPE=rabbitmq npm run dev:single would hit the same crash class. Default dev is Kafka and this card scopes the Kafka crash — worth a scope note / follow-up to mirror the fix on the RabbitMQ adapter.

Caveat: evo-flow CI runs only Sourcery — "23/23, typecheck clean, dev:single boots" is self-reported.

No blockers. Approving and merging (squash).