Skip to content

fix: per-topic Kafka builder methods drop SASL_SSL credentials from ConsumerConfig/ProducerConfig #3344

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jeremydmiller merged 1 commit into from
Jul 8, 2026
Merged

fix: per-topic Kafka builder methods drop SASL_SSL credentials from ConsumerConfig/ProducerConfig #3344

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
136 changes: 136 additions & 0 deletions src/Transports/Kafka/Wolverine.Kafka.Tests/KafkaTransportTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -246,6 +246,142 @@ public void begin_at_earliest_does_not_override_explicitly_set_group_id()
effective.GroupId.ShouldBe("topic-group");
effective.AutoOffsetReset.ShouldBe(AutoOffsetReset.Earliest);
}

// The methods below (BeginAtEarliest/BeginAtLatest/UseReadCommitted/UseCooperativeStickyAssignment/
// UseStaticMembership/TailFromLatest) all build a fresh per-topic ConsumerConfig containing only the
// one property they set, the same as BeginAtEarliest() did for GroupId before it was fixed above.
// GetEffectiveConsumerConfig() previously only backfilled BootstrapServers/GroupId from the parent,
// silently dropping SecurityProtocol/SaslMechanism/SaslUsername/SaslPassword -- a listener using any
// of these methods without a subsequent ExtendConsumerConfiguration() call would connect to a
// SASL_SSL broker without credentials and be disconnected during the initial handshake.

[Fact]
public void begin_at_earliest_inherits_sasl_ssl_settings_from_parent_transport()
{
var topic = BuildTopic();
topic.Parent.ConsumerConfig.SecurityProtocol = SecurityProtocol.SaslSsl;
topic.Parent.ConsumerConfig.SaslMechanism = SaslMechanism.Plain;
topic.Parent.ConsumerConfig.SaslUsername = "api-key";
topic.Parent.ConsumerConfig.SaslPassword = "api-secret";

var config = new KafkaListenerConfiguration(topic)
.BeginAtEarliest();
((IDelayedEndpointConfiguration)config).Apply();

var effective = topic.GetEffectiveConsumerConfig();
effective.SecurityProtocol.ShouldBe(SecurityProtocol.SaslSsl);
effective.SaslMechanism.ShouldBe(SaslMechanism.Plain);
effective.SaslUsername.ShouldBe("api-key");
effective.SaslPassword.ShouldBe("api-secret");
}

[Fact]
public void begin_at_latest_inherits_sasl_ssl_settings_from_parent_transport()
{
var topic = BuildTopic();
topic.Parent.ConsumerConfig.SecurityProtocol = SecurityProtocol.SaslSsl;
topic.Parent.ConsumerConfig.SaslMechanism = SaslMechanism.Plain;
topic.Parent.ConsumerConfig.SaslUsername = "api-key";
topic.Parent.ConsumerConfig.SaslPassword = "api-secret";

var config = new KafkaListenerConfiguration(topic)
.BeginAtLatest();
((IDelayedEndpointConfiguration)config).Apply();

var effective = topic.GetEffectiveConsumerConfig();
effective.SecurityProtocol.ShouldBe(SecurityProtocol.SaslSsl);
effective.SaslMechanism.ShouldBe(SaslMechanism.Plain);
effective.SaslUsername.ShouldBe("api-key");
effective.SaslPassword.ShouldBe("api-secret");
}

[Fact]
public void use_read_committed_inherits_sasl_ssl_settings_from_parent_transport()
{
var topic = BuildTopic();
topic.Parent.ConsumerConfig.SecurityProtocol = SecurityProtocol.SaslSsl;
topic.Parent.ConsumerConfig.SaslUsername = "api-key";
topic.Parent.ConsumerConfig.SaslPassword = "api-secret";

var config = new KafkaListenerConfiguration(topic)
.UseReadCommitted();
((IDelayedEndpointConfiguration)config).Apply();

var effective = topic.GetEffectiveConsumerConfig();
effective.SecurityProtocol.ShouldBe(SecurityProtocol.SaslSsl);
effective.SaslUsername.ShouldBe("api-key");
effective.SaslPassword.ShouldBe("api-secret");
}

[Fact]
public void use_cooperative_sticky_assignment_inherits_sasl_ssl_settings_from_parent_transport()
{
var topic = BuildTopic();
topic.Parent.ConsumerConfig.SecurityProtocol = SecurityProtocol.SaslSsl;
topic.Parent.ConsumerConfig.SaslUsername = "api-key";
topic.Parent.ConsumerConfig.SaslPassword = "api-secret";

var config = new KafkaListenerConfiguration(topic)
.UseCooperativeStickyAssignment();
((IDelayedEndpointConfiguration)config).Apply();

var effective = topic.GetEffectiveConsumerConfig();
effective.SecurityProtocol.ShouldBe(SecurityProtocol.SaslSsl);
effective.SaslUsername.ShouldBe("api-key");
effective.SaslPassword.ShouldBe("api-secret");
}

[Fact]
public void use_static_membership_inherits_sasl_ssl_settings_from_parent_transport()
{
var topic = BuildTopic();
topic.Parent.ConsumerConfig.SecurityProtocol = SecurityProtocol.SaslSsl;
topic.Parent.ConsumerConfig.SaslUsername = "api-key";
topic.Parent.ConsumerConfig.SaslPassword = "api-secret";

var config = new KafkaListenerConfiguration(topic)
.UseStaticMembership("node-1");
((IDelayedEndpointConfiguration)config).Apply();

var effective = topic.GetEffectiveConsumerConfig();
effective.SecurityProtocol.ShouldBe(SecurityProtocol.SaslSsl);
effective.SaslUsername.ShouldBe("api-key");
effective.SaslPassword.ShouldBe("api-secret");
}

[Fact]
public void tail_from_latest_inherits_sasl_ssl_settings_from_parent_transport()
{
var topic = BuildTopic();
topic.Parent.ConsumerConfig.SecurityProtocol = SecurityProtocol.SaslSsl;
topic.Parent.ConsumerConfig.SaslUsername = "api-key";
topic.Parent.ConsumerConfig.SaslPassword = "api-secret";

var config = new KafkaListenerConfiguration(topic)
.TailFromLatest();
((IDelayedEndpointConfiguration)config).Apply();

var effective = topic.GetEffectiveConsumerConfig();
effective.SecurityProtocol.ShouldBe(SecurityProtocol.SaslSsl);
effective.SaslUsername.ShouldBe("api-key");
effective.SaslPassword.ShouldBe("api-secret");
}

[Fact]
public void begin_at_earliest_does_not_override_explicitly_set_sasl_username()
{
var topic = BuildTopic();
topic.Parent.ConsumerConfig.SaslUsername = "parent-key";

var config = new KafkaListenerConfiguration(topic)
.ConfigureConsumer(c => c.SaslUsername = "topic-key")
.BeginAtEarliest();
((IDelayedEndpointConfiguration)config).Apply();

var effective = topic.GetEffectiveConsumerConfig();
effective.SaslUsername.ShouldBe("topic-key");
effective.AutoOffsetReset.ShouldBe(AutoOffsetReset.Earliest);
}
}

public class UseKafkaUsingNamedConnectionTests
Expand Down
42 changes: 42 additions & 0 deletions src/Transports/Kafka/Wolverine.Kafka.Tests/kafka_eos_building_blocks.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,48 @@ public void subscriber_idempotent_producer()
topic.ProducerConfig!.EnableIdempotence.ShouldBe(true);
}

[Fact]
public void subscriber_idempotent_producer_inherits_sasl_ssl_settings_from_parent_transport()
{
// UseIdempotentProducer() builds a fresh per-topic ProducerConfig containing only
// EnableIdempotence, the same pattern as the per-topic ConsumerConfig builders above.
// GetEffectiveProducerConfig() previously only backfilled BootstrapServers from the parent,
// silently dropping SecurityProtocol/SaslMechanism/SaslUsername/SaslPassword.
var transport = new KafkaTransport();
transport.ProducerConfig.SecurityProtocol = SecurityProtocol.SaslSsl;
transport.ProducerConfig.SaslMechanism = SaslMechanism.Plain;
transport.ProducerConfig.SaslUsername = "api-key";
transport.ProducerConfig.SaslPassword = "api-secret";

var topic = transport.Topics["events"];
var config = new KafkaSubscriberConfiguration(topic);
config.UseIdempotentProducer();
((IDelayedEndpointConfiguration)config).Apply();

var effective = topic.GetEffectiveProducerConfig();
effective.SecurityProtocol.ShouldBe(SecurityProtocol.SaslSsl);
effective.SaslMechanism.ShouldBe(SaslMechanism.Plain);
effective.SaslUsername.ShouldBe("api-key");
effective.SaslPassword.ShouldBe("api-secret");
}

[Fact]
public void subscriber_idempotent_producer_does_not_override_explicitly_set_sasl_username()
{
var transport = new KafkaTransport();
transport.ProducerConfig.SaslUsername = "parent-key";

var topic = transport.Topics["events"];
var config = new KafkaSubscriberConfiguration(topic);
config.ConfigureProducer(c => c.SaslUsername = "topic-key");
config.UseIdempotentProducer();
((IDelayedEndpointConfiguration)config).Apply();

var effective = topic.GetEffectiveProducerConfig();
effective.SaslUsername.ShouldBe("topic-key");
effective.EnableIdempotence.ShouldBe(true);
}

[Fact]
public void defaults_are_unchanged()
{
Expand Down
48 changes: 46 additions & 2 deletions src/Transports/Kafka/Wolverine.Kafka/KafkaTopic.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,7 +138,9 @@ public static string TopicNameForUri(Uri uri)
}

/// <summary>
/// Gets the effective ConsumerConfig for this topic, ensuring BootstrapServers is inherited from parent if not set
/// Gets the effective ConsumerConfig for this topic, ensuring BootstrapServers, GroupId, and the
/// security/SASL settings (SecurityProtocol, SaslMechanism, SaslUsername, SaslPassword) are
/// inherited from parent if not set
/// </summary>
internal ConsumerConfig GetEffectiveConsumerConfig()
{
Expand All @@ -152,11 +154,33 @@ internal ConsumerConfig GetEffectiveConsumerConfig()
ConsumerConfig.GroupId = Parent.ConsumerConfig.GroupId;
}

if (ConsumerConfig != null && ConsumerConfig.SecurityProtocol == null)
{
ConsumerConfig.SecurityProtocol = Parent.ConsumerConfig.SecurityProtocol;
}

if (ConsumerConfig != null && ConsumerConfig.SaslMechanism == null)
{
ConsumerConfig.SaslMechanism = Parent.ConsumerConfig.SaslMechanism;
}
Comment on lines +157 to +165

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

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

return ConsumerConfig ?? Parent.ConsumerConfig;
}

/// <summary>
/// Gets the effective ProducerConfig for this topic, ensuring BootstrapServers is inherited from parent if not set
/// Gets the effective ProducerConfig for this topic, ensuring BootstrapServers and the
/// security/SASL settings (SecurityProtocol, SaslMechanism, SaslUsername, SaslPassword) are
/// inherited from parent if not set
/// </summary>
internal ProducerConfig GetEffectiveProducerConfig()
{
Expand All @@ -165,6 +189,26 @@ internal ProducerConfig GetEffectiveProducerConfig()
ProducerConfig.BootstrapServers = Parent.ProducerConfig.BootstrapServers;
}

if (ProducerConfig != null && ProducerConfig.SecurityProtocol == null)
{
ProducerConfig.SecurityProtocol = Parent.ProducerConfig.SecurityProtocol;
}

if (ProducerConfig != null && ProducerConfig.SaslMechanism == null)
{
ProducerConfig.SaslMechanism = Parent.ProducerConfig.SaslMechanism;
}

if (ProducerConfig != null && string.IsNullOrEmpty(ProducerConfig.SaslUsername))
{
ProducerConfig.SaslUsername = Parent.ProducerConfig.SaslUsername;
}

if (ProducerConfig != null && string.IsNullOrEmpty(ProducerConfig.SaslPassword))
{
ProducerConfig.SaslPassword = Parent.ProducerConfig.SaslPassword;
}

return ProducerConfig ?? Parent.ProducerConfig;
}

Expand Down
Loading