feat(Kafka): Add KRaft support

src/Testcontainers.Kafka/ApacheConfiguration.cs

@@ -0,0 +1,62 @@
+namespace Testcontainers.Kafka;
+
+/// <inheritdoc cref="KafkaVendorConfiguration" />
+internal sealed class ApacheConfiguration : KafkaVendorConfiguration
+{
+    static ApacheConfiguration()
+    {
+    }
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="ApacheConfiguration" /> class.
+    /// </summary>
+    private ApacheConfiguration()
+    {
+    }
+
+    /// <summary>
+    /// Gets the singleton instance of the Apache vendor configuration.
+    /// </summary>
+    public static KafkaVendorConfiguration Instance { get; }
+        = new ApacheConfiguration();
+
+    /// <inheritdoc />
+    public override KafkaVendor Vendor
+        => KafkaVendor.ApacheSoftwareFoundation;
+
+    /// <inheritdoc />
+    public override ConsensusProtocol ConsensusProtocol
+        => ConsensusProtocol.KRaft;
+
+    /// <inheritdoc />
+    public override bool IsImageFromVendor(IImage image)
+    {
+        return image.Repository.Contains("apache") || image.Repository.Contains("bitnami");
+    }
+
+    /// <inheritdoc />
+    public override void Validate(KafkaConfiguration resourceConfiguration)
+    {
+        const string message = "Local ZooKeeper is not supported for Apache Kafka images. Configure an external ZooKeeper.";
+
+        var isZooKeeperConsensus = resourceConfiguration.ConsensusProtocol == ConsensusProtocol.ZooKeeper;
+
+        var hasLocalZooKeeper = isZooKeeperConsensus && resourceConfiguration.Environments.TryGetValue("KAFKA_ZOOKEEPER_CONNECT", out var connectionString) && connectionString.StartsWith("localhost");
+
+        _ = Guard.Argument(resourceConfiguration, nameof(IContainerConfiguration.Image))
+            .ThrowIf(_ => hasLocalZooKeeper, argument => new ArgumentException(message, argument.Name));
+    }
+
+    /// <inheritdoc />
+    public override string CreateStartupScript(KafkaConfiguration resourceConfiguration, KafkaContainer container)
+    {
+        var additionalAdvertisedListeners = string.Join(",", container.AdvertisedListeners ?? Array.Empty<string>());
+
+        var startupScript = new StringWriter();
+        startupScript.NewLine = "\n";
+        startupScript.WriteLine("#!/bin/bash");
+        startupScript.WriteLine("export KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://" + container.Hostname + ":" + container.GetMappedPublicPort(KafkaBuilder.KafkaPort) + ",BROKER://" + container.IpAddress + ":" + KafkaBuilder.BrokerPort + "," + additionalAdvertisedListeners);
+        startupScript.WriteLine("exec /etc/kafka/docker/run");
+        return startupScript.ToString();
+    }
+}

src/Testcontainers.Kafka/ConfluentConfiguration.cs

@@ -0,0 +1,75 @@
+namespace Testcontainers.Kafka;
+
+/// <inheritdoc cref="KafkaVendorConfiguration" />
+internal sealed class ConfluentConfiguration : KafkaVendorConfiguration
+{
+    static ConfluentConfiguration()
+    {
+    }
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="ConfluentConfiguration" /> class.
+    /// </summary>
+    private ConfluentConfiguration()
+    {
+    }
+
+    /// <summary>
+    /// Gets the singleton instance of the Confluent vendor configuration.
+    /// </summary>
+    public static KafkaVendorConfiguration Instance { get; }
+        = new ConfluentConfiguration();
+
+    /// <inheritdoc />
+    public override KafkaVendor Vendor
+        => KafkaVendor.Confluent;
+
+    /// <inheritdoc />
+    public override ConsensusProtocol ConsensusProtocol
+        => ConsensusProtocol.ZooKeeper;
+
+    /// <inheritdoc />
+    public override bool IsImageFromVendor(IImage image)
+    {
+        return image.Repository.Contains("confluentinc");
+    }
+
+    /// <inheritdoc />
+    public override void Validate(KafkaConfiguration resourceConfiguration)
+    {
+        const string message = "KRaft is not supported for Confluent Platform images with versions earlier than 7.0.0.";
+
+        Predicate<KafkaConfiguration> isUnsupportedImage = value => value.ConsensusProtocol == ConsensusProtocol.KRaft
+            && IsImageFromVendor(value.Image) && value.Image.MatchVersion(v => v.Major < 7);
+
+        _ = Guard.Argument(resourceConfiguration, nameof(IContainerConfiguration.Image))
+            .ThrowIf(argument => isUnsupportedImage(argument.Value), argument => new ArgumentException(message, argument.Name));
+    }
+
+    /// <inheritdoc />
+    public override string CreateStartupScript(KafkaConfiguration resourceConfiguration, KafkaContainer container)
+    {
+        var additionalAdvertisedListeners = string.Join(",", container.AdvertisedListeners ?? Array.Empty<string>());
+
+        var isZooKeeperConsensus = resourceConfiguration.ConsensusProtocol == ConsensusProtocol.ZooKeeper;
+
+        var hasLocalZooKeeper = isZooKeeperConsensus && resourceConfiguration.Environments.TryGetValue("KAFKA_ZOOKEEPER_CONNECT", out var connectionString) && connectionString.StartsWith("localhost");
+
+        var startupScript = new StringWriter();
+        startupScript.NewLine = "\n";
+        startupScript.WriteLine("#!/bin/bash");
+
+        if (isZooKeeperConsensus && hasLocalZooKeeper)
+        {
+            startupScript.WriteLine("echo '' > /etc/confluent/docker/ensure");
+            startupScript.WriteLine("echo 'clientPort=" + KafkaBuilder.ZooKeeperPort + "' > zookeeper.properties");
+            startupScript.WriteLine("echo 'dataDir=/var/lib/zookeeper/data' >> zookeeper.properties");
+            startupScript.WriteLine("echo 'dataLogDir=/var/lib/zookeeper/log' >> zookeeper.properties");
+            startupScript.WriteLine("zookeeper-server-start zookeeper.properties &");
+        }
+
+        startupScript.WriteLine("export KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://" + container.Hostname + ":" + container.GetMappedPublicPort(KafkaBuilder.KafkaPort) + ",BROKER://" + container.IpAddress + ":" + KafkaBuilder.BrokerPort + "," + additionalAdvertisedListeners);
+        startupScript.WriteLine("exec /etc/confluent/docker/run");
+        return startupScript.ToString();
+    }
+}