feat(kafka_native): new module

Author: strowk

.github/dependabot.yml

@@ -42,6 +42,7 @@ updates:
         - /modules/k3s
         - /modules/k6
         - /modules/kafka
+        - /modules/kafka_native
         - /modules/localstack
         - /modules/mariadb
         - /modules/meilisearch

docs/modules/kafka_native.md

@@ -0,0 +1,76 @@
+# Kafka Native
+
+Since <a href="https://github.com/testcontainers/testcontainers-go/releases/tag/v0.39.0"><span class="tc-version">:material-tag: v0.39.0</span></a>
+
+## Introduction
+
+The Testcontainers module for [Apache Kafka Native](https://hub.docker.com/r/apache/kafka-native).
+
+## Adding this module to your project dependencies
+
+Please run the following command to add the Kafka module to your Go dependencies:
+
+```
+go get github.com/testcontainers/testcontainers-go/modules/kafka_native
+```
+
+## Usage example
+
+<!--codeinclude-->
+[Creating a Kafka container](../../modules/kafka_native/examples_test.go) inside_block:runKafkaContainer
+<!--/codeinclude-->
+
+## Module Reference
+
+### Run function
+
+The Kafka module exposes one entrypoint function to create the Kafka container, and this function receives three parameters:
+
+```golang
+func Run(ctx context.Context, img string, opts ...testcontainers.ContainerCustomizer) (*KafkaContainer, error)
+```
+
+- `context.Context`, the Go context.
+- `string`, the Docker image to use.
+- `testcontainers.ContainerCustomizer`, a variadic argument for passing options.
+
+#### Image
+
+Use the second argument in the `Run` function to set a valid Docker image.
+In example: `Run(context.Background(), "apache/kafka-native:3.9.1")`.
+
+#### Environment variables
+
+The environment variables that are already set by default are:
+
+<!--codeinclude-->
+[Environment variables](../../modules/kafka_native/kafka.go) inside_block:envVars
+<!--/codeinclude-->
+
+And also KAFKA_ADVERTISED_LISTENERS that is defined dynamically based on the container's hostname.
+
+#### Init script
+
+The Kafka container will be started using a custom shell script:
+
+<!--codeinclude-->
+[Init script](../../modules/kafka_native/kafka.go) inside_block:starterScriptContentText
+<!--/codeinclude-->
+
+### Container Options
+
+When starting the Kafka container, you can pass options in a variadic way to configure it.
+
+{% include "../features/common_functional_options_list.md" %}
+
+### Container Methods
+
+The Kafka container exposes the following methods:
+
+#### Brokers
+
+The `Brokers(ctx)` method returns the Kafka brokers as a string slice, containing the host and the random port defined by Kafka's public port (`9093/tcp`).
+
+<!--codeinclude-->
+[Get Kafka brokers](../../modules/kafka_native/kafka_test.go) inside_block:getBrokers
+<!--/codeinclude-->

mkdocs.yml

@@ -93,6 +93,7 @@ nav:
         - modules/k3s.md
         - modules/k6.md
         - modules/kafka.md
+        - modules/kafka_native.md
         - modules/localstack.md
         - modules/mariadb.md
         - modules/meilisearch.md

modules/kafka_native/Makefile

@@ -0,0 +1,5 @@
+include ../../commons-test.mk
+
+.PHONY: test
+test:
+	$(MAKE) test-kafka-native

modules/kafka_native/consumer_test.go

@@ -0,0 +1,56 @@
+package kafka_native_test
+
+import (
+	"testing"
+
+	"github.com/IBM/sarama"
+)
+
+// TestKafkaConsumer is a test consumer for Kafka
+type TestKafkaConsumer struct {
+	t       *testing.T
+	ready   chan bool
+	done    chan bool
+	cancel  chan bool
+	message *sarama.ConsumerMessage
+}
+
+func NewTestKafkaConsumer(t *testing.T) (*TestKafkaConsumer, <-chan bool, <-chan bool, func()) {
+	t.Helper()
+	kc := &TestKafkaConsumer{
+		t:      t,
+		ready:  make(chan bool, 1),
+		done:   make(chan bool, 1),
+		cancel: make(chan bool, 1),
+	}
+	return kc, kc.ready, kc.done, func() {
+		kc.cancel <- true
+	}
+}
+
+func (k *TestKafkaConsumer) Setup(_ sarama.ConsumerGroupSession) error {
+	return nil
+}
+
+func (k *TestKafkaConsumer) Cleanup(_ sarama.ConsumerGroupSession) error {
+	return nil
+}
+
+// ConsumeClaim is called by the Kafka client library when a message is received
+func (k *TestKafkaConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+	k.ready <- true
+	for {
+		select {
+		case message := <-claim.Messages():
+			k.message = message
+			session.MarkMessage(message, "")
+			k.done <- true
+
+		case <-k.cancel:
+			return nil
+
+		case <-session.Context().Done():
+			return nil
+		}
+	}
+}

modules/kafka_native/examples_test.go

@@ -0,0 +1,43 @@
+package kafka_native_test
+
+import (
+	"context"
+	"fmt"
+	"log"
+
+	"github.com/testcontainers/testcontainers-go"
+	"github.com/testcontainers/testcontainers-go/modules/kafka_native"
+)
+
+func ExampleRun() {
+	// runKafkaContainer {
+	ctx := context.Background()
+
+	kafkaContainer, err := kafka_native.Run(ctx,
+		"apache/kafka-native:3.9.1",
+		kafka_native.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_native/go.mod

@@ -0,0 +1,85 @@
+module github.com/testcontainers/testcontainers-go/modules/kafka_native
+
+go 1.23.0
+
+toolchain go1.23.6
+
+require (
+	github.com/IBM/sarama v1.42.1
+	github.com/docker/go-connections v0.5.0
+	github.com/stretchr/testify v1.10.0
+	github.com/testcontainers/testcontainers-go v0.38.0
+	golang.org/x/mod v0.16.0
+)
+
+require (
+	dario.cat/mergo v1.0.1 // indirect
+	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
+	github.com/Microsoft/go-winio v0.6.2 // indirect
+	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
+	github.com/containerd/errdefs v1.0.0 // indirect
+	github.com/containerd/errdefs/pkg v0.3.0 // indirect
+	github.com/containerd/log v0.1.0 // indirect
+	github.com/containerd/platforms v0.2.1 // indirect
+	github.com/cpuguy83/dockercfg v0.3.2 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/distribution/reference v0.6.0 // indirect
+	github.com/docker/docker v28.2.2+incompatible // indirect
+	github.com/docker/go-units v0.5.0 // indirect
+	github.com/eapache/go-resiliency v1.4.0 // indirect
+	github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect
+	github.com/eapache/queue v1.1.0 // indirect
+	github.com/ebitengine/purego v0.8.4 // indirect
+	github.com/felixge/httpsnoop v1.0.4 // indirect
+	github.com/go-logr/logr v1.4.2 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-ole/go-ole v1.2.6 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/hashicorp/errwrap v1.1.0 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
+	github.com/hashicorp/go-uuid v1.0.3 // indirect
+	github.com/jcmturner/aescts/v2 v2.0.0 // indirect
+	github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect
+	github.com/jcmturner/gofork v1.7.6 // indirect
+	github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect
+	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
+	github.com/klauspost/compress v1.18.0 // indirect
+	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
+	github.com/magiconair/properties v1.8.10 // indirect
+	github.com/moby/docker-image-spec v1.3.1 // indirect
+	github.com/moby/go-archive v0.1.0 // indirect
+	github.com/moby/patternmatcher v0.6.0 // indirect
+	github.com/moby/sys/sequential v0.6.0 // indirect
+	github.com/moby/sys/user v0.4.0 // indirect
+	github.com/moby/sys/userns v0.1.0 // indirect
+	github.com/moby/term v0.5.0 // indirect
+	github.com/morikuni/aec v1.0.0 // indirect
+	github.com/opencontainers/go-digest v1.0.0 // indirect
+	github.com/opencontainers/image-spec v1.1.1 // indirect
+	github.com/pierrec/lz4/v4 v4.1.18 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
+	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
+	github.com/shirou/gopsutil/v4 v4.25.5 // indirect
+	github.com/sirupsen/logrus v1.9.3 // indirect
+	github.com/tklauser/go-sysconf v0.3.12 // indirect
+	github.com/tklauser/numcpus v0.6.1 // indirect
+	github.com/yusufpapurcu/wmi v1.2.4 // indirect
+	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0 // indirect
+	go.opentelemetry.io/otel v1.35.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.19.0 // indirect
+	go.opentelemetry.io/otel/metric v1.35.0 // indirect
+	go.opentelemetry.io/otel/trace v1.35.0 // indirect
+	go.opentelemetry.io/proto/otlp v1.0.0 // indirect
+	golang.org/x/crypto v0.37.0 // indirect
+	golang.org/x/net v0.38.0 // indirect
+	golang.org/x/sync v0.8.0 // indirect
+	golang.org/x/sys v0.32.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)
+
+replace github.com/testcontainers/testcontainers-go => ../..