feat(kafka): add apache/kafka and apache/kafka-native support

docs/modules/kafka.md

@@ -1,10 +1,12 @@
-# Kafka (KRaft)
+# Kafka
 
 Since <a href="https://github.com/testcontainers/testcontainers-go/releases/tag/v0.24.0"><span class="tc-version">:material-tag: v0.24.0</span></a>
 
 ## Introduction
 
-The Testcontainers module for KRaft: [Apache Kafka Without ZooKeeper](https://developer.confluent.io/learn/kraft).
+The Testcontainers module for Kafka.
+
+This module would run Kafka in Kraft mode: [Apache Kafka Without ZooKeeper](https://developer.confluent.io/learn/kraft/) and it supports both [Apache Kafka](https://kafka.apache.org/) and [Confluent](https://docs.confluent.io/kafka/overview.html) images.
 
 ## Adding this module to your project dependencies
 
@@ -17,9 +19,19 @@ go get github.com/testcontainers/testcontainers-go/modules/kafka
 ## Usage example
 
 <!--codeinclude-->
-[Creating a Kafka container](../../modules/kafka/examples_test.go) inside_block:runKafkaContainer
+[Apache Native Kafka](../../modules/kafka/examples_test.go) inside_block:runKafkaContainerApacheNative
+<!--/codeinclude-->
+
+<!--codeinclude-->
+[Apache Kafka](../../modules/kafka/examples_test.go) inside_block:runKafkaContainerApacheNotNative
+<!--/codeinclude-->
+
+<!--codeinclude-->
+[Confluent Kafka](../../modules/kafka/examples_test.go) inside_block:runKafkaContainerConfluent
 <!--/codeinclude-->
 
+The native container ([apache/kafka-native](https://hub.docker.com/r/apache/kafka-native/)) is based on GraalVM and typically starts several seconds faster than alternatives.
+
 ## Module Reference
 
 ### Run function
@@ -42,12 +54,12 @@ func Run(ctx context.Context, img string, opts ...testcontainers.ContainerCustom
 #### Image
 
 Use the second argument in the `Run` function to set a valid Docker image.
-In example: `Run(context.Background(), "confluentinc/confluent-local:7.5.0")`.
+In example: `Run(context.Background(), "apache/kafka-native:4.0.1")`.
 
 !!! warning
-    The minimal required version of Kafka for KRaft mode is `confluentinc/confluent-local:7.4.0`. If you are using an image that
-    is different from the official one, please make sure that it's compatible with KRaft mode, as the module won't check
-    the version for you.
+    Module expects that the image in use supports Kraft mode (Kafka without ZooKeeper).
+    The minimal required version of Confluent images for KRaft mode is `confluentinc/confluent-local:7.4.0`.
+    All Apache images support Kraft mode.
 
 #### Environment variables
 
@@ -59,10 +71,19 @@ The environment variables that are already set by default are:
 
 #### Init script
 
-The Kafka container will be started using a custom shell script:
+The Kafka container will be started using a custom shell script.
+
+Module would vary the starter script depending on the image in use, using following logic:
+
+- image starts with `apache/kafka`: use Apache Kafka starter script.
+- image starts with `confluentinc/`: use Confluent starter script.
+
+<!--codeinclude-->
+[Apache Kafka starter script](../../modules/kafka/kafka.go) inside_block:starterScriptApache
+<!--/codeinclude-->
 
 <!--codeinclude-->
-[Init script](../../modules/kafka/kafka.go) inside_block:starterScript
+[Confluent starter script](../../modules/kafka/kafka.go) inside_block:starterScriptConfluentinc
 <!--/codeinclude-->
 
 ### Container Options

modules/kafka/examples_test.go

@@ -9,8 +9,8 @@ import (
 	"github.com/testcontainers/testcontainers-go/modules/kafka"
 )
 
-func ExampleRun() {
-	// runKafkaContainer {
+func ExampleRun_confluentinc() {
+	// runKafkaContainerConfluentinc {
 	ctx := context.Background()
 
 	kafkaContainer, err := kafka.Run(ctx,
@@ -41,3 +41,69 @@ func ExampleRun() {
 	// test-cluster
 	// true
 }
+
+func ExampleRun_apacheNative() {
+	// runKafkaContainerApacheNative {
+	ctx := context.Background()
+
+	kafkaContainer, err := kafka.Run(ctx,
+		"apache/kafka-native:4.0.1",
+		kafka.WithClusterID("test-cluster"),
+	)
+	defer func() {
+		if err := testcontainers.TerminateContainer(kafkaContainer); err != nil {
+			log.Printf("failed to terminate container: %s", err)
+		}
+	}()
+	if err != nil {
+		log.Printf("failed to start container: %s", err)
+		return
+	}
+	// }
+
+	state, err := kafkaContainer.State(ctx)
+	if err != nil {
+		log.Printf("failed to get container state: %s", err)
+		return
+	}
+
+	fmt.Println(kafkaContainer.ClusterID)
+	fmt.Println(state.Running)
+
+	// Output:
+	// test-cluster
+	// true
+}
+
+func ExampleRun_apacheNotNative() {
+	// runKafkaContainerApacheNotNative {
+	ctx := context.Background()
+
+	kafkaContainer, err := kafka.Run(ctx,
+		"apache/kafka:4.0.1",
+		kafka.WithClusterID("test-cluster"),
+	)
+	defer func() {
+		if err := testcontainers.TerminateContainer(kafkaContainer); err != nil {
+			log.Printf("failed to terminate container: %s", err)
+		}
+	}()
+	if err != nil {
+		log.Printf("failed to start container: %s", err)
+		return
+	}
+	// }
+
+	state, err := kafkaContainer.State(ctx)
+	if err != nil {
+		log.Printf("failed to get container state: %s", err)
+		return
+	}
+
+	fmt.Println(kafkaContainer.ClusterID)
+	fmt.Println(state.Running)
+
+	// Output:
+	// test-cluster
+	// true
+}

modules/kafka/kafka.go

@@ -19,8 +19,8 @@ const publicPort = nat.Port("9093/tcp")
 const (
 	starterScript = "/usr/sbin/testcontainers_start.sh"
 
-	// starterScript {
-	starterScriptContent = `#!/bin/bash
+	// starterScriptConfluentinc {
+	confluentincStarterScriptContent = `#!/bin/bash
 source /etc/confluent/docker/bash-config
 export KAFKA_ADVERTISED_LISTENERS=%s,BROKER://%s:9092
 echo Starting Kafka KRaft mode
@@ -30,6 +30,13 @@ echo '' > /etc/confluent/docker/ensure
 /etc/confluent/docker/configure
 /etc/confluent/docker/launch`
 	// }
+
+	// starterScriptApache {
+	apacheStarterScriptContent = `#!/bin/bash
+export KAFKA_ADVERTISED_LISTENERS=%s,BROKER://%s:9092
+echo Starting Apache Kafka
+exec /etc/kafka/docker/run`
+	// }
 )
 
 // KafkaContainer represents the Kafka container type used in the module
@@ -78,7 +85,7 @@ func Run(ctx context.Context, img string, opts ...testcontainers.ContainerCustom
 					// if the starter script fails to copy.
 					func(ctx context.Context, c testcontainers.Container) error {
 						// 1. copy the starter script into the container
-						if err := copyStarterScript(ctx, c); err != nil {
+						if err := copyStarterScript(ctx, img, c); err != nil {
 							return fmt.Errorf("copy starter script: %w", err)
 						}
 
@@ -122,7 +129,7 @@ func Run(ctx context.Context, img string, opts ...testcontainers.ContainerCustom
 }
 
 // copyStarterScript copies the starter script into the container.
-func copyStarterScript(ctx context.Context, c testcontainers.Container) error {
+func copyStarterScript(ctx context.Context, img string, c testcontainers.Container) error {
 	if err := wait.ForMappedPort(publicPort).
 		WaitUntilReady(ctx, c); err != nil {
 		return fmt.Errorf("wait for mapped port: %w", err)
@@ -140,7 +147,7 @@ func copyStarterScript(ctx context.Context, c testcontainers.Container) error {
 
 	hostname := inspect.Config.Hostname
 
-	scriptContent := fmt.Sprintf(starterScriptContent, endpoint, hostname)
+	scriptContent := fmt.Sprintf(getStarterScriptContent(img), endpoint, hostname)
 
 	if err := c.CopyToContainer(ctx, []byte(scriptContent), starterScript, 0o755); err != nil {
 		return fmt.Errorf("copy to container: %w", err)
@@ -200,7 +207,7 @@ func validateKRaftVersion(fqName string) error {
 	image := fqName[:strings.LastIndex(fqName, ":")]
 	version := fqName[strings.LastIndex(fqName, ":")+1:]
 
-	if !strings.EqualFold(image, "confluentinc/confluent-local") {
+	if !isConfluentinc(image) {
 		// do not validate if the image is not the official one.
 		// not raising an error here, letting the image start and
 		// eventually evaluate an error if it exists.

modules/kafka/kafka_test.go

@@ -12,12 +12,12 @@ import (
 	"github.com/testcontainers/testcontainers-go/modules/kafka"
 )
 
-func TestKafka(t *testing.T) {
+func testFor(image string, t *testing.T) {
 	topic := "some-topic"
 
 	ctx := context.Background()
 
-	kafkaContainer, err := kafka.Run(ctx, "confluentinc/confluent-local:7.5.0", kafka.WithClusterID("kraftCluster"))
+	kafkaContainer, err := kafka.Run(ctx, image, kafka.WithClusterID("kraftCluster"))
 	testcontainers.CleanupContainer(t, kafkaContainer)
 	require.NoError(t, err)
 
@@ -66,6 +66,32 @@ func TestKafka(t *testing.T) {
 	require.Truef(t, strings.EqualFold(string(consumer.message.Value), "value"), "expected value to be %s, got %s", "value", string(consumer.message.Value))
 }
 
+func TestKafka(t *testing.T) {
+	testCases := []struct {
+		name  string
+		image string
+	}{
+		{
+			name:  "confluentinc",
+			image: "confluentinc/confluent-local:7.5.0",
+		},
+		{
+			name:  "apache native",
+			image: "apache/kafka-native:4.0.1",
+		},
+		{
+			name:  "apache not-native",
+			image: "apache/kafka:4.0.1",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			testFor(tc.image, t)
+		})
+	}
+}
+
 func TestKafka_invalidVersion(t *testing.T) {
 	ctx := context.Background()
 

modules/kafka/version.go

@@ -0,0 +1,29 @@
+package kafka
+
+import "strings"
+
+const (
+	apacheKafkaImagePrefix  = "apache/kafka"
+	confluentincImagePrefix = "confluentinc/"
+	dockerIoPrefix          = "docker.io/"
+)
+
+func isApache(image string) bool {
+	return strings.HasPrefix(image, apacheKafkaImagePrefix) || strings.HasPrefix(image, dockerIoPrefix+apacheKafkaImagePrefix)
+}
+
+func isConfluentinc(image string) bool {
+	return strings.HasPrefix(image, confluentincImagePrefix) || strings.HasPrefix(image, dockerIoPrefix+confluentincImagePrefix)
+}
+
+func getStarterScriptContent(image string) string {
+	if isApache(image) {
+		return apacheStarterScriptContent
+	} else if isConfluentinc(image) {
+		return confluentincStarterScriptContent
+	} else {
+		// Default to confluentinc for backward compatibility
+		// in situations when image was custom specified based on confluentinc
+		return confluentincStarterScriptContent
+	}
+}