fix(kafka): Fix internal docker connection

docs/modules/kafka.md

@@ -71,6 +71,44 @@ The environment variables that are already set by default are:
 
 {% include "../features/common_functional_options.md" %}
 
+
+#### ClusterId
+
+You can set up cluster id by using `WithClusterID` option.
+
+```
+KafkaContainer, err = kafka.Run(ctx,
+		"confluentinc/confluent-local:7.6.1",
+		kafka.WithClusterID("test-cluster"))
+```
+
+#### Listeners
+
+If you need to connect new listeners, you can use `WithListener(listeners []KafkaListener)`. 
+This option controls the following environment variables for the Kafka container: 
+- `KAFKA_LISTENERS`
+- `KAFKA_REST_BOOTSTRAP_SERVERS`
+- `KAFKA_LISTENER_SECURITY_PROTOCOL_MAP`
+- `KAFKA_INTER_BROKER_LISTENER_NAME`
+- `KAFKA_ADVERTISED_LISTENERS`
+
+Example:
+
+<!--codeinclude-->
+[Get Kafka brokers](../../modules/kafka/kafka_test.go) inside_block:kafkaWithListener
+<!--/codeinclude-->
+
+In the above code, we created a network for our container and attached kafka to it, so they can communicate. Then we marked port 9092 for our internal usage.
+
+The first listener in the slice will be written in the env parameter `KAFKA_INTER_BROKER_LISTENER_NAME`  
+
+Every listener's name will be converted in upper case. Every name and port should be unique and will be checked in a validation step.
+
+If you are not using this option or the listeners list is empty, there will be 2 default listeners with the following addresses and ports:
+
+External - Host():MappedPort()  
+Internal - Host():9092
+
 ### Container Methods
 
 The Kafka container exposes the following methods:
@@ -81,4 +119,4 @@ The `Brokers(ctx)` method returns the Kafka brokers as a string slice, containin
 
 <!--codeinclude-->
 [Get Kafka brokers](../../modules/kafka/kafka_test.go) inside_block:getBrokers
-<!--/codeinclude-->
+<!--/codeinclude-->

modules/kafka/go.mod

@@ -3,7 +3,7 @@ module github.com/testcontainers/testcontainers-go/modules/kafka
 go 1.22
 
 require (
-	github.com/IBM/sarama v1.42.1
+	github.com/IBM/sarama v1.43.3
 	github.com/docker/go-connections v0.5.0
 	github.com/stretchr/testify v1.9.0
 	github.com/testcontainers/testcontainers-go v0.34.0
@@ -23,7 +23,7 @@ require (
 	github.com/distribution/reference v0.6.0 // indirect
 	github.com/docker/docker v27.1.1+incompatible // indirect
 	github.com/docker/go-units v0.5.0 // indirect
-	github.com/eapache/go-resiliency v1.4.0 // indirect
+	github.com/eapache/go-resiliency v1.7.0 // indirect
 	github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect
 	github.com/eapache/queue v1.1.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
@@ -41,7 +41,7 @@ require (
 	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.17.4 // indirect
+	github.com/klauspost/compress v1.17.9 // indirect
 	github.com/kr/text v0.2.0 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
 	github.com/magiconair/properties v1.8.7 // indirect
@@ -53,7 +53,7 @@ require (
 	github.com/morikuni/aec v1.0.0 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/opencontainers/image-spec v1.1.0 // indirect
-	github.com/pierrec/lz4/v4 v4.1.18 // indirect
+	github.com/pierrec/lz4/v4 v4.1.21 // 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
@@ -69,8 +69,7 @@ require (
 	go.opentelemetry.io/otel/metric v1.24.0 // indirect
 	go.opentelemetry.io/otel/trace v1.24.0 // indirect
 	golang.org/x/crypto v0.28.0 // indirect
-	golang.org/x/net v0.26.0 // indirect
-	golang.org/x/sync v0.7.0 // indirect
+	golang.org/x/net v0.28.0 // indirect
 	golang.org/x/sys v0.26.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237 // indirect

modules/kafka/go.sum

@@ -4,8 +4,8 @@ github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24 h1:bvDV9
 github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8=
 github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 h1:UQHMgLO+TxOElx5B5HZ4hJQsoJ/PvUvKRhJHDQXO8P8=
 github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E=
-github.com/IBM/sarama v1.42.1 h1:wugyWa15TDEHh2kvq2gAy1IHLjEjuYOYgXz/ruC/OSQ=
-github.com/IBM/sarama v1.42.1/go.mod h1:Xxho9HkHd4K/MDUo/T/sOqwtX/17D33++E9Wib6hUdQ=
+github.com/IBM/sarama v1.43.3 h1:Yj6L2IaNvb2mRBop39N7mmJAHBVY3dTPncr3qGVkxPA=
+github.com/IBM/sarama v1.43.3/go.mod h1:FVIRaLrhK3Cla/9FfRF5X9Zua2KpS3SYIXxhac1H+FQ=
 github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY=
 github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU=
 github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM=
@@ -32,8 +32,8 @@ github.com/docker/go-connections v0.5.0 h1:USnMq7hx7gwdVZq1L49hLXaFtUdTADjXGp+uj
 github.com/docker/go-connections v0.5.0/go.mod h1:ov60Kzw0kKElRwhNs9UlUHAE/F9Fe6GLaXnqyDdmEXc=
 github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
 github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
-github.com/eapache/go-resiliency v1.4.0 h1:3OK9bWpPk5q6pbFAaYSEwD9CLUSHG8bnZuqX2yMt3B0=
-github.com/eapache/go-resiliency v1.4.0/go.mod h1:5yPzW0MIvSe0JDsv0v+DvcjEv2FyD6iZYSs1ZI+iQho=
+github.com/eapache/go-resiliency v1.7.0 h1:n3NRTnBn5N0Cbi/IeOHuQn9s2UwVUH7Ga0ZWcP+9JTA=
+github.com/eapache/go-resiliency v1.7.0/go.mod h1:5yPzW0MIvSe0JDsv0v+DvcjEv2FyD6iZYSs1ZI+iQho=
 github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 h1:Oy0F4ALJ04o5Qqpdz8XLIpNA3WM/iSIXqxtqo7UGVws=
 github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3/go.mod h1:YvSRo5mw33fLEx1+DlK6L2VV43tJt5Eyel9n9XBcR+0=
 github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc=
@@ -85,8 +85,8 @@ github.com/jcmturner/rpc/v2 v2.0.3 h1:7FXXj8Ti1IaVFpSAziCZWNzbNuZmnvw/i6CqLNdWfZ
 github.com/jcmturner/rpc/v2 v2.0.3/go.mod h1:VUJYCIDm3PVOEHw8sgt091/20OJjskO/YJki3ELg/Hc=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
-github.com/klauspost/compress v1.17.4 h1:Ej5ixsIri7BrIjBkRZLTo6ghwrEtHFk7ijlczPW4fZ4=
-github.com/klauspost/compress v1.17.4/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM=
+github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
+github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
 github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0=
 github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
@@ -111,8 +111,8 @@ github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8
 github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
 github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQb2IpWsCzug=
 github.com/opencontainers/image-spec v1.1.0/go.mod h1:W4s4sFTMaBeK1BQLXbG4AdM2szdn85PY75RI83NrTrM=
-github.com/pierrec/lz4/v4 v4.1.18 h1:xaKrnTkyoqfh1YItXl56+6KJNVYWlEEPuAQW9xsplYQ=
-github.com/pierrec/lz4/v4 v4.1.18/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
+github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ=
+github.com/pierrec/lz4/v4 v4.1.21/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -190,14 +190,14 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
-golang.org/x/net v0.26.0 h1:soB7SVo0PWrY4vPW/+ay0jKDNScG2X9wFeYlXIvJsOQ=
-golang.org/x/net v0.26.0/go.mod h1:5YKkiSynbBIh3p6iOc/vibscux0x38BZDkn8sCUPxHE=
+golang.org/x/net v0.28.0 h1:a9JDOJc5GMUJ0+UDqmLT86WiEy7iWyIhz8gz8E4e5hE=
+golang.org/x/net v0.28.0/go.mod h1:yqtgsTWOOnlGLG9GFRrK3++bGOUEkNBoHZc8MEDWPNg=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
-golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.8.0 h1:3NFvSEYkUoMifnESzZl15y791HH1qU2xm6eCJU5ZPXQ=
+golang.org/x/sync v0.8.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

modules/kafka/kafka.go

@@ -22,7 +22,7 @@ const (
 	// starterScript {
 	starterScriptContent = `#!/bin/bash
 source /etc/confluent/docker/bash-config
-export KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://%s:%d,BROKER://%s:9092
+export KAFKA_ADVERTISED_LISTENERS=%s
 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
@@ -36,6 +36,13 @@ echo '' > /etc/confluent/docker/ensure
 type KafkaContainer struct {
 	testcontainers.Container
 	ClusterID string
+	Listeners KafkaListener
+}
+
+type KafkaListener struct {
+	Name string
+	Host string
+	Port string
 }
 
 // Deprecated: use Run instead
@@ -70,37 +77,52 @@ func Run(ctx context.Context, img string, opts ...testcontainers.ContainerCustom
 		Entrypoint: []string{"sh"},
 		// this CMD will wait for the starter script to be copied into the container and then execute it
 		Cmd: []string{"-c", "while [ ! -f " + starterScript + " ]; do sleep 0.1; done; bash " + starterScript},
-		LifecycleHooks: []testcontainers.ContainerLifecycleHooks{
-			{
-				PostStarts: []testcontainers.ContainerHook{
-					// Use a single hook to copy the starter script and wait for
-					// the Kafka server to be ready. This prevents the wait running
-					// 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 {
-							return fmt.Errorf("copy starter script: %w", err)
-						}
-
-						// 2. wait for the Kafka server to be ready
-						return wait.ForLog(".*Transitioning from RECOVERY to RUNNING.*").AsRegexp().WaitUntilReady(ctx, c)
-					},
-				},
-			},
-		},
 	}
 
 	genericContainerReq := testcontainers.GenericContainerRequest{
 		ContainerRequest: req,
 		Started:          true,
 	}
 
+	settings := defaultOptions()
 	for _, opt := range opts {
+		if apply, ok := opt.(Option); ok {
+			apply(&settings)
+		}
 		if err := opt.Customize(&genericContainerReq); err != nil {
 			return nil, err
 		}
 	}
 
+	if err := trimValidateListeners(settings.Listeners); err != nil {
+		return nil, fmt.Errorf("listeners validation: %w", err)
+	}
+
+	// apply envs for listeners
+	envChange := editEnvsForListeners(settings.Listeners)
+	for key, item := range envChange {
+		genericContainerReq.Env[key] = item
+	}
+
+	genericContainerReq.ContainerRequest.LifecycleHooks = []testcontainers.ContainerLifecycleHooks{
+		{
+			PostStarts: []testcontainers.ContainerHook{
+				// Use a single hook to copy the starter script and wait for
+				// the Kafka server to be ready. This prevents the wait running
+				// 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, &settings); err != nil {
+						return fmt.Errorf("copy starter script: %w", err)
+					}
+
+					// 2. wait for the Kafka server to be ready
+					return wait.ForLog(".*Transitioning from RECOVERY to RUNNING.*").AsRegexp().WaitUntilReady(ctx, c)
+				},
+			},
+		},
+	}
+
 	err := validateKRaftVersion(genericContainerReq.Image)
 	if err != nil {
 		return nil, err
@@ -121,32 +143,70 @@ func Run(ctx context.Context, img string, opts ...testcontainers.ContainerCustom
 	return c, nil
 }
 
+func trimValidateListeners(listeners []KafkaListener) error {
+	// Trim
+	for i := 0; i < len(listeners); i++ {
+		listeners[i].Name = strings.ToUpper(strings.Trim(listeners[i].Name, " "))
+		listeners[i].Host = strings.Trim(listeners[i].Host, " ")
+		listeners[i].Port = strings.Trim(listeners[i].Port, " ")
+	}
+
+	// Validate
+	var ports map[string]bool = make(map[string]bool, len(listeners)+2)
+	var names map[string]bool = make(map[string]bool, len(listeners)+2)
+
+	// check for default listeners
+	ports["9094"] = true
+	ports["9093"] = true
+
+	// check for default listeners
+	names["CONTROLLER"] = true
+	names["PLAINTEXT"] = true
+
+	for _, item := range listeners {
+		if names[item.Name] {
+			return fmt.Errorf("duplicate of listener name: %s", item.Name)
+		}
+		names[item.Name] = true
+
+		if ports[item.Port] {
+			return fmt.Errorf("duplicate of listener port: %s", item.Port)
+		}
+		ports[item.Port] = true
+	}
+
+	return nil
+}
+
 // copyStarterScript copies the starter script into the container.
-func copyStarterScript(ctx context.Context, c testcontainers.Container) error {
+func copyStarterScript(ctx context.Context, c testcontainers.Container, settings *options) error {
 	if err := wait.ForListeningPort(publicPort).
 		SkipInternalCheck().
 		WaitUntilReady(ctx, c); err != nil {
 		return fmt.Errorf("wait for exposed port: %w", err)
 	}
 
-	host, err := c.Host(ctx)
-	if err != nil {
-		return fmt.Errorf("host: %w", err)
+	if len(settings.Listeners) == 0 {
+		defaultInternal, err := brokerListener(ctx, c)
+		if err != nil {
+			return fmt.Errorf("can't create default internal listener: %w", err)
+		}
+		settings.Listeners = append(settings.Listeners, defaultInternal)
 	}
 
-	inspect, err := c.Inspect(ctx)
+	defaultExternal, err := plainTextListener(ctx, c)
 	if err != nil {
-		return fmt.Errorf("inspect: %w", err)
+		return fmt.Errorf("can't create default external listener: %w", err)
 	}
 
-	hostname := inspect.Config.Hostname
+	settings.Listeners = append(settings.Listeners, defaultExternal)
 
-	port, err := c.MappedPort(ctx, publicPort)
-	if err != nil {
-		return fmt.Errorf("mapped port: %w", err)
+	var advertised []string
+	for _, item := range settings.Listeners {
+		advertised = append(advertised, fmt.Sprintf("%s://%s:%s", item.Name, item.Host, item.Port))
 	}
 
-	scriptContent := fmt.Sprintf(starterScriptContent, host, port.Int(), hostname)
+	scriptContent := fmt.Sprintf(starterScriptContent, strings.Join(advertised, ","))
 
 	if err := c.CopyToContainer(ctx, []byte(scriptContent), starterScript, 0o755); err != nil {
 		return fmt.Errorf("copy to container: %w", err)
@@ -155,12 +215,43 @@ func copyStarterScript(ctx context.Context, c testcontainers.Container) error {
 	return nil
 }
 
-func WithClusterID(clusterID string) testcontainers.CustomizeRequestOption {
-	return func(req *testcontainers.GenericContainerRequest) error {
-		req.Env["CLUSTER_ID"] = clusterID
+func editEnvsForListeners(listeners []KafkaListener) map[string]string {
+	if len(listeners) == 0 {
+		// no change
+		return map[string]string{}
+	}
 
-		return nil
+	envs := map[string]string{
+		"KAFKA_LISTENERS":                      "CONTROLLER://0.0.0.0:9094, PLAINTEXT://0.0.0.0:9093",
+		"KAFKA_REST_BOOTSTRAP_SERVERS":         "CONTROLLER://0.0.0.0:9094, PLAINTEXT://0.0.0.0:9093",
+		"KAFKA_LISTENER_SECURITY_PROTOCOL_MAP": "CONTROLLER:PLAINTEXT, PLAINTEXT:PLAINTEXT",
+	}
+
+	// expect first listener has common network between kafka instances
+	envs["KAFKA_INTER_BROKER_LISTENER_NAME"] = listeners[0].Name
+
+	// expect small number of listeners, so joins is okay
+	for _, item := range listeners {
+		envs["KAFKA_LISTENERS"] = strings.Join(
+			[]string{
+				envs["KAFKA_LISTENERS"],
+				fmt.Sprintf("%s://0.0.0.0:%s", item.Name, item.Port),
+			},
+			",",
+		)
+
+		envs["KAFKA_REST_BOOTSTRAP_SERVERS"] = envs["KAFKA_LISTENERS"]
+
+		envs["KAFKA_LISTENER_SECURITY_PROTOCOL_MAP"] = strings.Join(
+			[]string{
+				envs["KAFKA_LISTENER_SECURITY_PROTOCOL_MAP"],
+				item.Name + ":" + "PLAINTEXT",
+			},
+			",",
+		)
 	}
+
+	return envs
 }
 
 // Brokers retrieves the broker connection strings from Kafka with only one entry,