fix: BeginAtEarliest/BeginAtLatest drops GroupId from transport-level ConsumerConfig

[lahma]

Bug

KafkaListenerConfiguration.BeginAtEarliest() / BeginAtLatest() call:

topic.ConsumerConfig ??= new ConsumerConfig();
topic.ConsumerConfig.AutoOffsetReset = AutoOffsetReset.Earliest;

This creates a fresh per-topic ConsumerConfig with only AutoOffsetReset set.
KafkaTopic.GetEffectiveConsumerConfig() then returns this sparse config — it already
copied BootstrapServers from the parent, but not GroupId. Confluent.Kafka requires
group.id and throws at startup:

System.ArgumentException: 'group.id' configuration parameter is required and was not specified.

This is a breaking change introduced with the BeginAtEarliest() / BeginAtLatest()
methods added in 6.13.x (GH-3146).

Root cause

GetEffectiveConsumerConfig() in KafkaTopic.cs:

internal ConsumerConfig GetEffectiveConsumerConfig()
{
    if (ConsumerConfig != null && string.IsNullOrEmpty(ConsumerConfig.BootstrapServers))
    {
        ConsumerConfig.BootstrapServers = Parent.ConsumerConfig.BootstrapServers;
    }
    // GroupId was never propagated — the returned config has no group.id
    return ConsumerConfig ?? Parent.ConsumerConfig;
}

Fix

Add GroupId inheritance immediately after BootstrapServers, using the same pattern:

if (ConsumerConfig != null && string.IsNullOrEmpty(ConsumerConfig.GroupId))
{
    ConsumerConfig.GroupId = Parent.ConsumerConfig.GroupId;
}

Tests

Three new unit tests in KafkaListenerConfigurationTests:

• begin_at_earliest_inherits_group_id_from_parent_transport — verifies GroupId + AutoOffsetReset.Earliest + BootstrapServers are all present after .BeginAtEarliest()
• begin_at_latest_inherits_group_id_from_parent_transport — same for .BeginAtLatest()
• begin_at_earliest_does_not_override_explicitly_set_group_id — verifies a per-topic GroupId set via ConfigureConsumer is not overwritten

Workaround (for users who cannot wait for a fix release)

Use ExtendConsumerConfiguration instead of BeginAtEarliest():

opts.ListenToKafkaTopic("my-topic")
    .ExtendConsumerConfiguration(c => c.AutoOffsetReset = AutoOffsetReset.Earliest);

[Copilot]

Pull request overview

This PR fixes a regression in the Kafka transport where calling KafkaListenerConfiguration.BeginAtEarliest() / BeginAtLatest() could result in a per-topic ConsumerConfig missing GroupId, causing Confluent.Kafka startup failures (group.id is required).

Changes:

• Update KafkaTopic.GetEffectiveConsumerConfig() to inherit GroupId from the parent transport when a topic-level ConsumerConfig exists but has no GroupId.
• Add unit tests covering BeginAtEarliest() / BeginAtLatest() inheritance of GroupId and ensuring an explicitly configured per-topic GroupId is not overridden.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated no comments.

File	Description
src/Transports/Kafka/Wolverine.Kafka/KafkaTopic.cs	Ensures topic-level consumer configs inherit GroupId from the parent transport when missing.
src/Transports/Kafka/Wolverine.Kafka.Tests/KafkaTransportTests.cs	Adds regression tests validating BeginAtEarliest/Latest behavior and GroupId inheritance semantics.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.