How to setup Kafka container with SaSl Configuration

[vermapraveen]

can someone suggest on how achieve setup kafka testcontainer with sasl. I am trying to write INT test around existing consumer which connect kafka using Sasl. Producer/ consumer in test is not able to communicate with kafka with sasl. Producer and Consumer work ok without Sasl, so no issue from Testcontainer local setup.
*This is my first time dealing with Sasl

Fixture

namespace IntegrationTests
{
    using System;
    using System.Threading.Tasks;
    using DotNet.Testcontainers.Builders;
    using DotNet.Testcontainers.Configurations;
    using DotNet.Testcontainers.Containers;
    using Xunit;

    public sealed class KafkaFixture : IAsyncLifetime, IDisposable
    {
        public static readonly string Topic = "Test";
        public static readonly string GroupId = "Test_group";
        public static readonly string UserName = "Test_user";
        public static readonly string Password = "Test_pwd";
        private readonly KafkaTestcontainerConfiguration configuration = new KafkaTestcontainerConfiguration { };

        public KafkaFixture()
        {
            configuration.Environments.Add("KAFKA_SASL_ENABLED_MECHANISMS", "PLAIN");
            configuration.Environments.Add("KAFKA_LISTENER_NAME_PLAINTEXT_SASL_ENABLED_MECHANISMS", "PLAIN");
            configuration.Environments.Add("KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL", "PLAIN");

            this.Kafka = new TestcontainersBuilder<KafkaTestcontainer>()
              .WithKafka(this.configuration)
              .Build();
        }

        public KafkaTestcontainer Kafka { get; }

        public Task InitializeAsync()
        {
            return this.Kafka.StartAsync();
        }

        public Task DisposeAsync()
        {
            return this.Kafka.DisposeAsync().AsTask();
        }

        public void Dispose()
        {
            this.configuration.Dispose();
        }
    }
}

Producer:

        private static async Task Producer(Message<string, string> message, string server, string topic)
        {
            var deliveryReportTaskCompletionSource = new TaskCompletionSource<DeliveryReport<string, string>>();
            var producerConfig = new ProducerConfig
            {
                BootstrapServers = server,
                SaslUsername = KafkaFixture.UserName,
                SaslPassword = KafkaFixture.Password,
                SaslMechanism = SaslMechanism.Plain,
                EnableSslCertificateVerification = false,
                SecurityProtocol = SecurityProtocol.SaslSsl,
            };
            using var producer = new ProducerBuilder<string, string>(producerConfig).Build();
            producer.Produce(topic, message, deliveryHandler => deliveryReportTaskCompletionSource.SetResult(deliveryHandler));
            await deliveryReportTaskCompletionSource.Task;
        }

Consumer:

        private static string Consumer(string server, string topic)
        {
            var config = new ConsumerConfig
            {
                BootstrapServers = server,
                GroupId = KafkaFixture.GroupId,
                SaslUsername = KafkaFixture.UserName,
                SaslPassword = KafkaFixture.Password,
                SaslMechanism = SaslMechanism.Plain,
                EnableSslCertificateVerification = false,
                SecurityProtocol = SecurityProtocol.SaslSsl,
                AutoOffsetReset = AutoOffsetReset.Latest,
            };

            using var consumer = new ConsumerBuilder<string, string>(config).Build();
            consumer.Subscribe(topic);

            var result = consumer.Consume(TimeSpan.FromSeconds(5));
            return result.Message.Value;
        }

Below is test log after I aborted the test

The active test run was aborted. Reason: Test host process crashed
========== Test run aborted: 0 Tests (0 Passed, 0 Failed, 0 Skipped) run in < 1 ms ==========
Building Test Projects
========== Starting test run ==========
[xUnit.net 00:00:00.01] xUnit.net VSTest Adapter v2.4.5+1caef2f33e (64-bit .NET Core 3.1.29)
[xUnit.net 00:00:02.91]   Starting:    IntegrationTests
The active test run was aborted. Reason: Test host process crashed : %3|1665767015.415|FAIL|rdkafka#producer-1| [thrd:sasl_ssl://localhost:49160/bootstrap]: sasl_ssl://localhost:49160/bootstrap: Connect to ipv6#[::1]:49160 failed: Unknown error (after 2021ms in state CONNECT)
%3|1665767015.415|ERROR|rdkafka#producer-1| [thrd:sasl_ssl://localhost:49160/bootstrap]: 1/1 brokers are down
%3|1665767015.415|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: sasl_ssl://localhost:49160/bootstrap: Connect to ipv6#[::1]:49160 failed: Unknown error (after 2021ms in state CONNECT)
%3|1665767018.449|FAIL|rdkafka#producer-1| [thrd:sasl_ssl://localhost:49160/bootstrap]: sasl_ssl://localhost:49160/bootstrap: Connect to ipv4#127.0.0.1:49160 failed: Unknown error (after 2023ms in state CONNECT)
%3|1665767018.449|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: sasl_ssl://localhost:49160/bootstrap: Connect to ipv4#127.0.0.1:49160 failed: Unknown error (after 2023ms in state CONNECT)
%3|1665767021.455|FAIL|rdkafka#producer-1| [thrd:sasl_ssl://localhost:49160/bootstrap]: sasl_ssl://localhost:49160/bootstrap: Connect to ipv4#127.0.0.1:49160 failed: Unknown error (after 2018ms in state CONNECT, 1 identical error(s) suppressed)
%3|1665767021.455|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: sasl_ssl://localhost:49160/bootstrap: Connect to ipv4#127.0.0.1:49160 failed: Unknown error (after 2018ms in state CONNECT, 1 identical error(s) suppressed)
%3|1665767024.491|FAIL|rdkafka#producer-1| [thrd:sasl_ssl://localhost:49160/bootstrap]: sasl_ssl://localhost:49160/bootstrap: Connect to ipv6#[::1]:49160 failed: Unknown error (after 2033ms in state CONNECT)
%3|1665767024.491|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: sasl_ssl://localhost:49160/bootstrap: Connect to ipv6#[::1]:49160 failed: Unknown error (after 2033ms in state CONNECT)
%3|1665767030.575|FAIL|rdkafka#producer-1| [thrd:sasl_ssl://localhost:49160/bootstrap]: sasl_ssl://localhost:49160/bootstrap: Connect to ipv4#127.0.0.1:49160 failed: Unknown error (after 2061ms in state CONNECT)
%3|1665767030.575|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: sasl_ssl://localhost:49160/bootstrap: Connect to ipv4#127.0.0.1:49160 failed: Unknown error (after 2061ms in state CONNECT)
%3|1665767033.569|FAIL|rdkafka#producer-1| [thrd:sasl_ssl://localhost:49160/bootstrap]: sasl_ssl://localhost:49160/bootstrap: Connect to ipv6#[::1]:49160 failed: Unknown error (after 2042ms in state CONNECT)
%3|1665767033.569|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: sasl_ssl://localhost:49160/bootstrap: Connect to ipv6#[::1]:49160 failed: Unknown error (after 2042ms in state CONNECT)
%3|1665767039.625|FAIL|rdkafka#producer-1| [thrd:sasl_ssl://localhost:49160/bootstrap]: sasl_ssl://localhost:49160/bootstrap: Connect to ipv4#127.0.0.1:49160 failed: Unknown error (after 2030ms in state CONNECT)
%3|1665767039.625|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: sasl_ssl://localhost:49160/bootstrap: Connect to ipv4#127.0.0.1:49160 failed: Unknown error (after 2030ms in state CONNECT)
%3|1665767042.670|FAIL|rdkafka#producer-1| [thrd:sasl_ssl://localhost:49160/bootstrap]: sasl_ssl://localhost:49160/bootstrap: Connect to ipv6#[::1]:49160 failed: Unknown error (after 2047ms in state CONNECT)
%3|1665767042.670|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: sasl_ssl://localhost:49160/bootstrap: Connect to ipv6#[::1]:49160 failed: Unknown error (after 2047ms in state CONNECT)
%3|1665767048.720|FAIL|rdkafka#producer-1| [thrd:sasl_ssl://localhost:49160/bootstrap]: sasl_ssl://localhost:49160/bootstrap: Connect to ipv4#127.0.0.1:49160 failed: Unknown error (after 2057ms in state CONNECT)
%3|1665767048.720|ERROR|rdkafka#producer-1| [thrd:app]: rdkafka#producer-1: sasl_ssl://localhost:49160/bootstrap: Connect to ipv4#127.0.0.1:49160 failed: Unknown error (after 2057ms in state CONNECT)
%3|1665767051.734|FAIL|rdkafka#producer-1| [thrd:sasl_ssl://localhost:49160/bootstrap]: sasl_ssl://localhost:49160/bootstrap: Connect to ipv4#127.0.0.1:49160 failed: Unknown error (after 2043ms in state CONNECT, 1 identical error(s) suppressed)

[kiview]

Hi @vermapraveen, unfortunately, this is a topic I don't have any experience with and the instructions around it seem quite complex:

• https://docs.confluent.io/platform/current/kafka/authentication_sasl/index.html
• https://docs.vmware.com/en/VMware-Smart-Assurance/10.1.0/sa-ui-installation-config-guide-10.1.0/GUID-3E473EC3-732A-4963-81BD-13BCCD3AC700.html
• https://docs.vmware.com/en/VMware-Smart-Assurance/10.1.0/sa-ui-installation-config-guide-10.1.0/GUID-DF659094-60D3-4E1B-8D63-3DE3ED8B0EDF.html

My suggestion right now: Try a manual Docker-based setup of the system, and once you have it working, we can support in replicating it with Testcontainers.