diff --git a/docs/modules/kafka.md b/docs/modules/kafka.md index 87c0020b04..8896c73648 100644 --- a/docs/modules/kafka.md +++ b/docs/modules/kafka.md @@ -1,10 +1,12 @@ -# Kafka (KRaft) +# Kafka Since :material-tag: v0.24.0 ## Introduction -The Testcontainers module for KRaft: [Apache Kafka Without ZooKeeper](https://developer.confluent.io/learn/kraft). +The Testcontainers module for Kafka. + +This module runs Kafka in Kraft mode: [Apache Kafka Without ZooKeeper](https://developer.confluent.io/learn/kraft/). ## Adding this module to your project dependencies @@ -14,10 +16,39 @@ Please run the following command to add the Kafka module to your Go dependencies go get github.com/testcontainers/testcontainers-go/modules/kafka ``` +## Apache Kafka images + +- Not available until the next release :material-tag: main + +Images `apache/kafka`, `apache/kafka-native` ([Apache Kafka](https://kafka.apache.org/)) are supported by this module in addition to `confluentinc/confluent-local` ([Confluent](https://docs.confluent.io/kafka/overview.html)). + +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. + +It is recommended to prefer Apache Kafka images over Confluent images, as Confluent has [unresolved issue with graceful shutdown](https://github.com/testcontainers/testcontainers-go/issues/2206). + +Apache Kafka Native images are also smallest, however they do not include CLI tools such as `kafka-topics.sh`. + +| Docker Image | Size | Start/stop time | CLI Tools | Graceful Shutdown | +|---------------------|--------------------------|-----------------|-----------|-------------------| +| Apache Kafka Native | 137MB (4.0.1 linux amd) | <1 second | No | OK | +| Apache Kafka | 393MB (4.0.1 linux amd) | ~3-4 seconds | Yes | OK | +| Confluent Kafka | 649MB (7.5.0 linux amd) | ~13-14 seconds | Yes | [issue](https://github.com/testcontainers/testcontainers-go/issues/2206) | + +!!!info + If you use image from custom registry, you might need to override starter script, see [Starter script](#starter-script) section below. + ## Usage example -[Creating a Kafka container](../../modules/kafka/examples_test.go) inside_block:runKafkaContainer +[Apache Kafka Native](../../modules/kafka/examples_test.go) inside_block:runKafkaContainerApacheNative + + + +[Apache Kafka](../../modules/kafka/examples_test.go) inside_block:runKafkaContainerApacheNotNative + + + +[Confluent Kafka](../../modules/kafka/examples_test.go) inside_block:runKafkaContainerConfluentinc ## Module Reference @@ -42,12 +73,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 @@ -57,18 +88,58 @@ The environment variables that are already set by default are: [Environment variables](../../modules/kafka/kafka.go) inside_block:envVars -#### Init script +#### Starter 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: -The Kafka container will be started using a custom shell script: +- image starts with `apache/kafka` or `docker.io/apache/kafka`: use Apache Kafka starter script. +- image starts with `confluentinc/` or `docker.io/confluentinc/`: use Confluent starter script. +- otherwise: use Confluent starter script (for backward compatibility). + +See also [WithApacheFlavor/WithConfluentFlavor](#withapacheflavorwithconfluentflavor) and [WithStarterScript](#withstarterscript) options to override this behavior. + + +[Apache Kafka starter script](../../modules/kafka/kafka.go) inside_block:starterScriptApache + -[Init script](../../modules/kafka/kafka.go) inside_block:starterScript +[Confluent starter script](../../modules/kafka/kafka.go) inside_block:starterScriptConfluentinc + + + +[Overriding starter script](../../modules/kafka/examples_test.go) inside_block:runKafkaContainerWithApacheFlavor ### Container Options When starting the Kafka container, you can pass options in a variadic way to configure it. +#### WithApacheFlavor/WithConfluentFlavor + +- Not available until the next release :material-tag: main + +You can manually specify which flavor of starter script to use with the following options: + + +[With Apache Flavor](../../modules/kafka/examples_test.go) inside_block:runKafkaContainerWithApacheFlavor + + + +[With Confluent Flavor](../../modules/kafka/examples_test.go) inside_block:runKafkaContainerWithConfluentFlavor + + +Note that both `WithApacheFlavor` and `WithConfluentFlavor` conflict with each other and with `WithStarterScript` option. An error will be returned if several of those options are provided. + +#### WithStarterScript + +- Not available until the next release :material-tag: main + +This allows to provide a completely custom starter script for the Kafka container. Be careful when using this option, as compatibility with any image and module version cannot be guaranteed. + +Note that `WithStarterScript` conflicts with `WithApacheFlavor` and `WithConfluentFlavor` options. An error will be returned if several of those options are provided. + {% include "../features/common_functional_options_list.md" %} ### Container Methods @@ -84,3 +155,17 @@ The `Brokers(ctx)` method returns the Kafka brokers as a string slice, containin [Get Kafka brokers](../../modules/kafka/kafka_test.go) inside_block:getBrokers + +#### Localhost listener + +- Not available until the next release :material-tag: main + +Kafka container would by default be configured with `localhost:9095` as one of advertised listeners. This can be used when you need to run CLI commands inside the container, for example with custom wait strategies or to prepare test data. + +Here is an example that uses custom wait strategy that checks if listing topics works: + + +[Custom wait strategy](../../modules/kafka/examples_test.go) inside_block:runKafkaContainerAndUseLocalhostListener + + +Note: this will not work with `apache/kafka-native` images, as they do not include CLI tools. \ No newline at end of file diff --git a/modules/kafka/benchmark_test.go b/modules/kafka/benchmark_test.go new file mode 100644 index 0000000000..29913c127f --- /dev/null +++ b/modules/kafka/benchmark_test.go @@ -0,0 +1,36 @@ +package kafka_test + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/testcontainers/testcontainers-go" + "github.com/testcontainers/testcontainers-go/modules/kafka" +) + +func startStopBenchmark(b *testing.B, image string) { + b.Helper() + for b.Loop() { + kafkaContainer, err := kafka.Run(context.Background(), + image, + ) + require.NoError(b, err) + + err = testcontainers.TerminateContainer(kafkaContainer) + require.NoError(b, err) + } +} + +func BenchmarkConfluentStartStop(b *testing.B) { + startStopBenchmark(b, "confluentinc/confluent-local:7.5.0") +} + +func BenchmarkApacheNativeStartStop(b *testing.B) { + startStopBenchmark(b, "apache/kafka-native:4.0.1") +} + +func BenchmarkApacheStartStop(b *testing.B) { + startStopBenchmark(b, "apache/kafka:4.0.1") +} diff --git a/modules/kafka/examples_test.go b/modules/kafka/examples_test.go index c275924ecc..cfd7f86343 100644 --- a/modules/kafka/examples_test.go +++ b/modules/kafka/examples_test.go @@ -4,13 +4,15 @@ import ( "context" "fmt" "log" + "time" "github.com/testcontainers/testcontainers-go" "github.com/testcontainers/testcontainers-go/modules/kafka" + "github.com/testcontainers/testcontainers-go/wait" ) -func ExampleRun() { - // runKafkaContainer { +func ExampleRun_confluentinc() { + // runKafkaContainerConfluentinc { ctx := context.Background() kafkaContainer, err := kafka.Run(ctx, @@ -41,3 +43,192 @@ 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 +} + +func ExampleRun_apacheNative_withApacheFlavor() { + // runKafkaContainerWithApacheFlavor { + ctx := context.Background() + + kafkaContainer, err := kafka.Run(ctx, + // the image might be different, for example + // custom-registry/apache/kafka-native:4.0.1, + // in which case the starter script would not + // be correctly inferred, and should be overridden + "apache/kafka-native:4.0.1", + kafka.WithClusterID("test-cluster"), + // this explicitly sets the starter script to use + // the one compatible with Apache images + kafka.WithApacheFlavor(), + ) + // } + 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_confluentinc_withConfluentFlavor() { + // runKafkaContainerWithConfluentFlavor { + ctx := context.Background() + + kafkaContainer, err := kafka.Run(ctx, + // the image might be different, for example + // custom-registry/confluentinc/confluent-local:7.5.0, + // in which case the starter script might not + // be correctly inferred, and should be overridden + "confluentinc/confluent-local:7.5.0", + kafka.WithClusterID("test-cluster"), + // this explicitly sets the starter script to use + // the one compatible with Confluent images + kafka.WithConfluentFlavor(), + ) + // } + 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_usingLocalhostListener() { + ctx := context.Background() + + // runKafkaContainerAndUseLocalhostListener { + kafkaContainer, err := kafka.Run(ctx, "apache/kafka:4.0.1", + testcontainers.WithWaitStrategy( + wait.NewExecStrategy([]string{ + "/opt/kafka/bin/kafka-topics.sh", + "--bootstrap-server", + "localhost:9095", + "--list", + }). + WithExitCode(0). + WithPollInterval(2*time.Second). + WithStartupTimeout(120*time.Second), + ), + 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 +} diff --git a/modules/kafka/kafka.go b/modules/kafka/kafka.go index ebbb6b987d..253ca201d4 100644 --- a/modules/kafka/kafka.go +++ b/modules/kafka/kafka.go @@ -17,12 +17,12 @@ import ( const publicPort = nat.Port("9093/tcp") const ( - starterScript = "/usr/sbin/testcontainers_start.sh" + starterScriptPath = "/usr/sbin/testcontainers_start.sh" - // starterScript { - starterScriptContent = `#!/bin/bash + // starterScriptConfluentinc { + confluentStarterScript = `#!/bin/bash source /etc/confluent/docker/bash-config -export KAFKA_ADVERTISED_LISTENERS=%s,BROKER://%s:9092 +export KAFKA_ADVERTISED_LISTENERS=%s,BROKER://%s:9092,LOCALHOST://localhost:9095 echo Starting Kafka KRaft mode sed -i '/KAFKA_ZOOKEEPER_CONNECT/d' /etc/confluent/docker/configure echo 'kafka-storage format --ignore-formatted -t "$(kafka-storage random-uuid)" -c /etc/kafka/kafka.properties' >> /etc/confluent/docker/configure @@ -30,6 +30,13 @@ echo '' > /etc/confluent/docker/ensure /etc/confluent/docker/configure /etc/confluent/docker/launch` // } + + // starterScriptApache { + apacheStarterScript = `#!/bin/bash +export KAFKA_ADVERTISED_LISTENERS=%s,BROKER://%s:9092,LOCALHOST://localhost:9095 +echo Starting Apache Kafka +exec /etc/kafka/docker/run` + // } ) // KafkaContainer represents the Kafka container type used in the module @@ -50,13 +57,24 @@ func Run(ctx context.Context, img string, opts ...testcontainers.ContainerCustom return nil, err } + runOptions := runOptions{ + image: img, + } + for _, opt := range opts { + if apply, ok := opt.(RunOption); ok { + if err := apply(&runOptions); err != nil { + return nil, fmt.Errorf("apply option: %w", err) + } + } + } + moduleOpts := []testcontainers.ContainerCustomizer{ testcontainers.WithExposedPorts(string(publicPort)), testcontainers.WithEnv(map[string]string{ // envVars { - "KAFKA_LISTENERS": "PLAINTEXT://0.0.0.0:9093,BROKER://0.0.0.0:9092,CONTROLLER://0.0.0.0:9094", - "KAFKA_REST_BOOTSTRAP_SERVERS": "PLAINTEXT://0.0.0.0:9093,BROKER://0.0.0.0:9092,CONTROLLER://0.0.0.0:9094", - "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP": "BROKER:PLAINTEXT,PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT", + "KAFKA_LISTENERS": "PLAINTEXT://0.0.0.0:9093,BROKER://0.0.0.0:9092,CONTROLLER://0.0.0.0:9094,LOCALHOST://localhost:9095", + "KAFKA_REST_BOOTSTRAP_SERVERS": "PLAINTEXT://0.0.0.0:9093,BROKER://0.0.0.0:9092,CONTROLLER://0.0.0.0:9094,LOCALHOST://localhost:9095", + "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP": "BROKER:PLAINTEXT,PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,LOCALHOST:PLAINTEXT", "KAFKA_INTER_BROKER_LISTENER_NAME": "BROKER", "KAFKA_BROKER_ID": "1", "KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR": "1", @@ -72,7 +90,7 @@ func Run(ctx context.Context, img string, opts ...testcontainers.ContainerCustom }), testcontainers.WithEntrypoint("sh"), // this CMD will wait for the starter script to be copied into the container and then execute it - testcontainers.WithCmd("-c", "while [ ! -f "+starterScript+" ]; do sleep 0.1; done; bash "+starterScript), + testcontainers.WithCmd("-c", "while [ ! -f "+starterScriptPath+" ]; do sleep 0.1; done; bash "+starterScriptPath), testcontainers.WithLifecycleHooks(testcontainers.ContainerLifecycleHooks{ PostStarts: []testcontainers.ContainerHook{ // Use a single hook to copy the starter script and wait for @@ -80,7 +98,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, &runOptions, c); err != nil { return fmt.Errorf("copy starter script: %w", err) } @@ -122,7 +140,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, opts *runOptions, c testcontainers.Container) error { if err := wait.ForMappedPort(publicPort). WaitUntilReady(ctx, c); err != nil { return fmt.Errorf("wait for mapped port: %w", err) @@ -140,21 +158,15 @@ func copyStarterScript(ctx context.Context, c testcontainers.Container) error { hostname := inspect.Config.Hostname - scriptContent := fmt.Sprintf(starterScriptContent, endpoint, hostname) + scriptContent := fmt.Sprintf(opts.getStarterScriptContent(), endpoint, hostname) - if err := c.CopyToContainer(ctx, []byte(scriptContent), starterScript, 0o755); err != nil { + if err := c.CopyToContainer(ctx, []byte(scriptContent), starterScriptPath, 0o755); err != nil { return fmt.Errorf("copy to container: %w", err) } return nil } -func WithClusterID(clusterID string) testcontainers.CustomizeRequestOption { - return testcontainers.WithEnv(map[string]string{ - "CLUSTER_ID": clusterID, - }) -} - // Brokers retrieves the broker connection strings from Kafka with only one entry, // defined by the exposed public port. func (kc *KafkaContainer) Brokers(ctx context.Context) ([]string, error) { @@ -207,7 +219,7 @@ func validateKRaftVersion(fqName string) error { image := fqName[:idx] version := fqName[idx+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. diff --git a/modules/kafka/kafka_test.go b/modules/kafka/kafka_test.go index af858f849f..9ba8acbe6c 100644 --- a/modules/kafka/kafka_test.go +++ b/modules/kafka/kafka_test.go @@ -4,20 +4,24 @@ import ( "context" "strings" "testing" + "time" "github.com/IBM/sarama" "github.com/stretchr/testify/require" "github.com/testcontainers/testcontainers-go" "github.com/testcontainers/testcontainers-go/modules/kafka" + "github.com/testcontainers/testcontainers-go/wait" ) -func TestKafka(t *testing.T) { +func testFor(t *testing.T, image string) { + t.Helper() + 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 +70,44 @@ 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 7.4.0", + image: "confluentinc/confluent-local:7.4.0", + }, + { + name: "confluentinc 7.5.0", + image: "confluentinc/confluent-local:7.5.0", + }, + { + name: "apache native 4", + image: "apache/kafka-native:4.0.1", + }, + { + name: "apache not-native 4", + image: "apache/kafka:4.0.1", + }, + { + name: "apache native 3.9", + image: "apache/kafka-native:3.9.1", + }, + { + name: "apache not-native 3.9", + image: "apache/kafka:3.9.1", + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + testFor(t, tc.image) + }) + } +} + func TestKafka_invalidVersion(t *testing.T) { ctx := context.Background() @@ -89,3 +131,104 @@ func assertAdvertisedListeners(t *testing.T, container testcontainers.Container) require.Containsf(t, bs, brokerURL, "expected advertised listeners to contain %s, got %s", brokerURL, bs) } + +func TestKafkaGracefulShutdown(t *testing.T) { + testCases := []struct { + name string + image string + }{ + { + 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) { + ctx := context.Background() + kafkaContainer, err := kafka.Run(ctx, tc.image) + testcontainers.CleanupContainer(t, kafkaContainer) + require.NoError(t, err) + + done := make(chan struct{}) + var stopErr error + go func() { + stopTimeout := 120 * time.Second + stopErr = kafkaContainer.Stop(ctx, &stopTimeout) + close(done) + }() + gracefulShutdownTimeout := 60 * time.Second + select { + case <-done: + require.NoError(t, stopErr) + case <-time.After(gracefulShutdownTimeout): + require.Failf(t, "Kafka did not gracefully exit", "Kafka did not gracefully exit in %v", gracefulShutdownTimeout) + } + }) + } +} + +func TestKafkaLocalhostListener(t *testing.T) { + testCases := []struct { + name string + image string + topicsExecPath string + }{ + { + name: "confluentinc 7.5.0", + image: "confluentinc/confluent-local:7.5.0", + topicsExecPath: "/bin/kafka-topics", + }, + { + name: "apache 4", + image: "apache/kafka:4.0.1", + topicsExecPath: "/opt/kafka/bin/kafka-topics.sh", + }, + // Note: this will not work for native images, because they do not include command line tools + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + + kafkaContainer, err := kafka.Run(ctx, tc.image, + testcontainers.WithWaitStrategy( + wait.NewExecStrategy([]string{ + tc.topicsExecPath, + "--bootstrap-server", + "localhost:9095", + "--list", + }). + WithExitCode(0). + WithPollInterval(2*time.Second). + WithStartupTimeout(120*time.Second))) + testcontainers.CleanupContainer(t, kafkaContainer, testcontainers.StopTimeout(0)) + require.NoError(t, err) + }) + } +} + +func TestFailOnBothFlavors(t *testing.T) { + ctx := context.Background() + + _, err := kafka.Run(ctx, "apache/kafka-native:4.0.1", + kafka.WithApacheFlavor(), + kafka.WithConfluentFlavor(), + ) + require.Error(t, err) + + _, err = kafka.Run(ctx, "apache/kafka-native:4.0.1", + kafka.WithApacheFlavor(), + kafka.WithStarterScript("testscript"), + ) + require.Error(t, err) + + _, err = kafka.Run(ctx, "apache/kafka-native:4.0.1", + kafka.WithConfluentFlavor(), + kafka.WithStarterScript("testscript"), + ) + require.Error(t, err) +} diff --git a/modules/kafka/options.go b/modules/kafka/options.go new file mode 100644 index 0000000000..39f0573b9d --- /dev/null +++ b/modules/kafka/options.go @@ -0,0 +1,90 @@ +package kafka + +import ( + "errors" + + "github.com/testcontainers/testcontainers-go" +) + +type runOptions struct { + image string + starterScript string + flavorWasSet bool +} + +// RunOption is an option that configures how Kafka container is started. +type RunOption func(*runOptions) error + +var _ testcontainers.ContainerCustomizer = (RunOption)(nil) + +func (o RunOption) Customize(_ *testcontainers.GenericContainerRequest) error { + return nil +} + +// WithStarterScript is an option to set a custom starter script content for the Kafka container. +// +// You would typically use this option when the image you are using is different from +// the standard ones and the default starter script does not work as expected. +// This option conflicts with WithApacheFlavor and WithConfluentFlavor options, +// and the error is returned if several are provided. +func WithStarterScript(content string) RunOption { + return func(o *runOptions) error { + if o.flavorWasSet { + return errFlavorAlreadySet + } + o.flavorWasSet = true + o.starterScript = content + return nil + } +} + +func (o *runOptions) getStarterScriptContent() string { + if o.starterScript == "" { + if isApache(o.image) { + return apacheStarterScript + } + // Default to confluentinc for backward compatibility + // in situations when image was custom specified based on confluentinc + return confluentStarterScript + } + return o.starterScript +} + +// WithClusterID sets the CLUSTER_ID environment variable for the Kafka container. +func WithClusterID(clusterID string) testcontainers.CustomizeRequestOption { + return testcontainers.WithEnv(map[string]string{ + "CLUSTER_ID": clusterID, + }) +} + +var errFlavorAlreadySet = errors.New("flavor was already set, provide only one of WithApacheFlavor, WithConfluentFlavor or WithStarterScript") + +// WithApacheFlavor sets the starter script to the one compatible with Apache Kafka images. +// +// Note: this option conflicts with WithConfluentFlavor and WithStarterScript options, +// and the error is returned if several are provided. +func WithApacheFlavor() RunOption { + return func(o *runOptions) error { + if o.flavorWasSet { + return errFlavorAlreadySet + } + o.flavorWasSet = true + o.starterScript = apacheStarterScript + return nil + } +} + +// WithConfluentFlavor sets the starter script to the one compatible with Confluent Kafka images. +// +// Note: this option conflicts with WithApacheFlavor and WithStarterScript options, +// and the error is returned if several are provided. +func WithConfluentFlavor() RunOption { + return func(o *runOptions) error { + if o.flavorWasSet { + return errFlavorAlreadySet + } + o.flavorWasSet = true + o.starterScript = confluentStarterScript + return nil + } +} diff --git a/modules/kafka/options_test.go b/modules/kafka/options_test.go new file mode 100644 index 0000000000..6ecd7b7623 --- /dev/null +++ b/modules/kafka/options_test.go @@ -0,0 +1,81 @@ +package kafka + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func Test_runOptions_getStarterScriptContent(t *testing.T) { + tests := []struct { + name string + image string + want string + }{ + { + name: "apache native image - latest", + image: "apache/kafka-native:latest", + want: apacheStarterScript, + }, + { + name: "apache native image - specific version", + image: "apache/kafka-native:4.0.1", + want: apacheStarterScript, + }, + { + name: "apache native image - specific version with docker.io prefix", + image: "docker.io/apache/kafka-native:4.0.1", + want: apacheStarterScript, + }, + { + name: "apache non-native image - latest", + image: "apache/kafka:latest", + want: apacheStarterScript, + }, + { + name: "apache non-native image - specific version", + image: "apache/kafka:4.0.0", + want: apacheStarterScript, + }, + { + name: "apache non-native image - with docker.io prefix", + image: "docker.io/apache/kafka:4.0.0", + want: apacheStarterScript, + }, + { + name: "confluentinc image - latest", + image: "confluentinc/cp-kafka:latest", + want: confluentStarterScript, + }, + { + name: "confluentinc image - no tag", + image: "confluentinc/cp-kafka", + want: confluentStarterScript, + }, + { + name: "confluentinc image - specific version", + image: "confluentinc/cp-kafka:8.1.0", + want: confluentStarterScript, + }, + { + name: "confluentinc image - specific version with docker.io prefix", + image: "docker.io/confluentinc/cp-kafka:8.1.0", + want: confluentStarterScript, + }, + { + name: "custom image", + image: "custom/kafka:latest", + want: confluentStarterScript, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + opts := &runOptions{ + image: tt.image, + } + require.Equal(t, tt.want, opts.getStarterScriptContent()) + require.NoError(t, WithStarterScript("mytestcript")(opts)) + require.Equal(t, "mytestcript", opts.getStarterScriptContent()) + }) + } +} diff --git a/modules/kafka/version.go b/modules/kafka/version.go new file mode 100644 index 0000000000..3c2c05558d --- /dev/null +++ b/modules/kafka/version.go @@ -0,0 +1,17 @@ +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) +} diff --git a/modules/kafka/version_test.go b/modules/kafka/version_test.go new file mode 100644 index 0000000000..c91961426a --- /dev/null +++ b/modules/kafka/version_test.go @@ -0,0 +1,115 @@ +package kafka + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func Test_isApache(t *testing.T) { + tests := []struct { + name string + image string + want bool + }{ + { + name: "apache native image - no tag", + image: "apache/kafka-native", + want: true, + }, + { + name: "apache native image - latest", + image: "apache/kafka-native:latest", + want: true, + }, + { + name: "apache native image - specific version", + image: "apache/kafka-native:4.0.1", + want: true, + }, + { + name: "apache native image - specific version with docker.io prefix", + image: "docker.io/apache/kafka-native:4.0.1", + want: true, + }, + { + name: "apache not-native image - no tag", + image: "apache/kafka", + want: true, + }, + { + name: "apache not-native image - latest", + image: "apache/kafka:latest", + want: true, + }, + { + name: "apache not-native image - specific version", + image: "apache/kafka:4.0.1", + want: true, + }, + { + name: "apache not-native image - specific version with docker.io prefix", + image: "docker.io/apache/kafka:4.0.1", + want: true, + }, + { + name: "confluentinc image", + image: "confluentinc/cp-kafka:latest", + want: false, + }, + { + name: "custom image", + image: "custom/kafka:latest", + want: false, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + require.Equal(t, tt.want, isApache(tt.image)) + }) + } +} + +func Test_isConfluentinc(t *testing.T) { + tests := []struct { + name string + image string + want bool + }{ + { + name: "confluentinc image - no tag", + image: "confluentinc/cp-kafka", + want: true, + }, + { + name: "confluentinc image - latest", + image: "confluentinc/cp-kafka:latest", + want: true, + }, + { + name: "confluentinc image - specific version", + image: "confluentinc/cp-kafka:8.1.0", + want: true, + }, + { + name: "confluentinc image - specific version with docker.io prefix", + image: "docker.io/confluentinc/cp-kafka:8.1.0", + want: true, + }, + { + name: "apache native image", + image: "apache/kafka-native:latest", + want: false, + }, + { + name: "custom image", + image: "custom/kafka:latest", + want: false, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + require.Equal(t, tt.want, isConfluentinc(tt.image)) + }) + } +}