feat(kafka): allow setting replay group id

src/Transports/Kafka/Wolverine.Kafka.Tests/kafka_replay_envelope_and_group.cs

@@ -106,6 +106,16 @@ public void replay_group_takes_the_prefix_of_the_topic_consumer_group()
             .ShouldStartWith("tenant-a-orders-replay-");
     }
 
+    [Fact]
+    public void replay_uses_a_requested_group_id_verbatim()
+    {
+        var (options, transport) = buildTransport(transportGroupId: "tenant-a");
+        var topic = transport.Topics["orders"];
+
+        KafkaReplay.ReplayGroupIdFor(topic, transport, options.ServiceName, "tenant-a-replay")
+            .ShouldBe("tenant-a-replay");
+    }
+
     [Fact]
     public void replay_group_takes_the_prefix_of_a_topic_group_subscribing_to_the_topic()
     {

src/Transports/Kafka/Wolverine.Kafka/Internals/KafkaReplay.cs

@@ -7,10 +7,10 @@
 namespace Wolverine.Kafka.Internals;
 
 /// <summary>
-/// Executes a bounded, one-shot Kafka replay (GH-3147): a separate <c>Assign()</c>-based consumer (with a
-/// unique throwaway group id and no commits) seeks to the requested start on each partition, reads
-/// forward to the end boundary, and feeds every record through the normal Wolverine handler pipeline. The
-/// live consumer group's committed offsets are never touched.
+/// Executes a bounded, one-shot Kafka replay (GH-3147): a separate <c>Assign()</c>-based consumer (with
+/// either the requested group id or a unique throwaway group id, and no commits) seeks to the requested
+/// start on each partition, reads forward to the end boundary, and feeds every record through the normal
+/// Wolverine handler pipeline. The live consumer group's committed offsets are never touched.
 /// </summary>
 internal sealed class KafkaReplay
 {
@@ -45,11 +45,11 @@ public async Task<KafkaReplayResult> ExecuteAsync(KafkaReplayRequest request, Ca
         // actually listens to this topic, which is not always the topic itself.
         var liveEndpoint = LiveListeningEndpointFor(topic, _transport);
 
-        // Throwaway, Assign-only consumer: unique group sharing the live group's prefix, no commits, no
-        // offset store.
+        // Assign-only consumer: use the requested group id when configured, otherwise a unique group sharing the
+        // live group's prefix. It never commits or stores offsets.
         var config = new ConsumerConfig(_transport.ConsumerConfig)
         {
-            GroupId = ReplayGroupIdFor(topic, _transport, _runtime.Options.ServiceName),
+            GroupId = ReplayGroupIdFor(topic, _transport, _runtime.Options.ServiceName, request.GroupId),
             EnableAutoCommit = false,
             EnableAutoOffsetStore = false
         };
@@ -146,12 +146,18 @@ public async Task<KafkaReplayResult> ExecuteAsync(KafkaReplayRequest request, Ca
     /// the service name. Brokers commonly grant consumer groups by prefix (Confluent Cloud ACLs are the
     /// usual case), and the live group is the one prefix the application is known to hold; the service
     /// name alone defaults to the entry assembly name, which such an ACL refuses with "Group
-    /// authorization failed".
+    /// authorization failed". A requested group id overrides this generated id verbatim.
     ///
     /// <para>A hot-tail listener's group is deliberately not borrowed; see <see cref="liveGroupIdOf" />.</para>
     /// </summary>
-    internal static string ReplayGroupIdFor(KafkaTopic topic, KafkaTransport transport, string serviceName)
+    internal static string ReplayGroupIdFor(KafkaTopic topic, KafkaTransport transport, string serviceName,
+        string? requestedGroupId = null)
     {
+        if (!string.IsNullOrWhiteSpace(requestedGroupId))
+        {
+            return requestedGroupId;
+        }
+
         var liveGroupId = liveGroupIdOf(LiveListeningEndpointFor(topic, transport));
 
         if (string.IsNullOrEmpty(liveGroupId))

src/Transports/Kafka/Wolverine.Kafka/KafkaReplayRequest.cs

@@ -13,6 +13,12 @@ public class KafkaReplayRequest
     /// <summary>The topic to replay.</summary>
     public required string Topic { get; init; }
 
+    /// <summary>
+    /// The consumer group id to use for this replay. Set this when the broker authorizes only specific group ids.
+    /// When omitted, the replay derives a unique id from the topic's live consumer group as before.
+    /// </summary>
+    public string? GroupId { get; init; }
+
     /// <summary>Start the replay at this absolute offset on every partition. Mutually exclusive with <see cref="FromTimestamp"/>.</summary>
     public long? FromOffset { get; init; }