chore: use confluent kafka image for tests (#289)

Replace `docker.redpanda.com/redpandadata/redpanda` for
`confluentinc/confluent-local` docker image for Kafka tests.

The sarama tests are currently flaking with the following error:

```
    ibm_sarama.go:127: 
        	Error Trace:	/home/runner/work/orchestrion/orchestrion/_integration-tests/tests/ibm_sarama/ibm_sarama.go:87
        	            				/home/runner/work/orchestrion/orchestrion/_integration-tests/tests/ibm_sarama/ibm_sarama.go:127
        	            				/home/runner/work/orchestrion/orchestrion/_integration-tests/utils/suite.go:77
        	            				/home/runner/work/orchestrion/orchestrion/_integration-tests/tests/ibm_sarama/gen_test.go:18
        	Error:      	Received unexpected error:
        	            	kafka: client has run out of available brokers to talk to: kafka: error decoding packet: message of length 1213486160 too large or too small
        	Test:       	TestIntegration_ibm_sarama
        	Messages:   	failed to create producer
```

We suspect it might be caused by the test starting before the container
being fully ready. The confluent kafka testcontainers module seems to
have smarter [readiness check
logic](https://github.com/testcontainers/testcontainers-go/blob/main/modules/kafka/kafka.go#L71-L89)
than the
[redpanda](https://github.com/testcontainers/testcontainers-go/blob/main/modules/redpanda/redpanda.go#L89-L95)
one, so hopefully this change solves the issue.

_integration-tests/go.mod

@@ -34,9 +34,9 @@ require (
 	github.com/testcontainers/testcontainers-go v0.33.0
 	github.com/testcontainers/testcontainers-go/modules/cassandra v0.33.0
 	github.com/testcontainers/testcontainers-go/modules/gcloud v0.33.0
+	github.com/testcontainers/testcontainers-go/modules/kafka v0.33.0
 	github.com/testcontainers/testcontainers-go/modules/mongodb v0.32.0
 	github.com/testcontainers/testcontainers-go/modules/redis v0.33.0
-	github.com/testcontainers/testcontainers-go/modules/redpanda v0.33.0
 	github.com/testcontainers/testcontainers-go/modules/vault v0.32.0
 	github.com/xlab/treeprint v1.2.0
 	go.mongodb.org/mongo-driver v1.16.1

_integration-tests/go.sum

@@ -1366,8 +1366,6 @@ github.com/mattn/go-sqlite3 v1.14.0/go.mod h1:JIl7NbARA7phWnGvh0LKTyg7S9BA+6gx71
 github.com/mattn/go-sqlite3 v1.14.14/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
 github.com/mattn/go-sqlite3 v1.14.22 h1:2gZY6PC6kBnID23Tichd1K+Z0oS6nE/XwU+Vz/5o4kU=
 github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
-github.com/mdelapenya/tlscert v0.1.0 h1:YTpF579PYUX475eOL+6zyEO3ngLTOUWck78NBuJVXaM=
-github.com/mdelapenya/tlscert v0.1.0/go.mod h1:wrbyM/DwbFCeCeqdPX/8c6hNOqQgbf0rUDErE1uD+64=
 github.com/microsoft/go-mssqldb v0.21.0 h1:p2rpHIL7TlSv1QrbXJUAcbyRKnIT0C9rRkH2E4OjLn8=
 github.com/microsoft/go-mssqldb v0.21.0/go.mod h1:+4wZTUnz/SV6nffv+RRRB/ss8jPng5Sho2SmM1l2ts4=
 github.com/minio/asm2plan9s v0.0.0-20200509001527-cdd76441f9d8/go.mod h1:mC1jAcsrzbxHt8iiaC+zU4b1ylILSosueou12R++wfY=
@@ -1571,12 +1569,12 @@ github.com/testcontainers/testcontainers-go/modules/cassandra v0.33.0 h1:Bpgl1+K
 github.com/testcontainers/testcontainers-go/modules/cassandra v0.33.0/go.mod h1:N6PuXZPZSUPioP8u5X67M8rPScL852hpZSMm8TDD3Tg=
 github.com/testcontainers/testcontainers-go/modules/gcloud v0.33.0 h1:0dLRqvSBSlWNZyJKRg9z0RjbnD2mF5CvzQi0M6a7oWM=
 github.com/testcontainers/testcontainers-go/modules/gcloud v0.33.0/go.mod h1:Lws7N+bAoAWEECrUZrku5T2T7i4GNj7UaVi1k/wZznU=
+github.com/testcontainers/testcontainers-go/modules/kafka v0.33.0 h1:Zug/9wK9tE9NdJ/sy27XbIvIVjXsD7rXrcV5B3KVrOM=
+github.com/testcontainers/testcontainers-go/modules/kafka v0.33.0/go.mod h1:J8NhxBCTnivcTANoNpuMnOEkdKf/x9zwZ434Y4dbIH4=
 github.com/testcontainers/testcontainers-go/modules/mongodb v0.32.0 h1:DvmvHV1irfNIVBhixeTAcoaWCvmdkoNQxRmZisqic4E=
 github.com/testcontainers/testcontainers-go/modules/mongodb v0.32.0/go.mod h1:z0ZvM2V2iThZGrzEN6sddJpvnGhJd6O1O0FTFoZXmpk=
 github.com/testcontainers/testcontainers-go/modules/redis v0.33.0 h1:S/QvMOwpr00MM2aWH+krzP73Erlp/Ug0dr2rkgZYI5s=
 github.com/testcontainers/testcontainers-go/modules/redis v0.33.0/go.mod h1:gudb3+6uZ9SsAysOVoLs7nazbjGlkHegBW8nqPXvDMI=
-github.com/testcontainers/testcontainers-go/modules/redpanda v0.33.0 h1:VscyHm3YRhU7MeaDFTCkesifv4Ff9NfqCrrQA056JkI=
-github.com/testcontainers/testcontainers-go/modules/redpanda v0.33.0/go.mod h1:VLRAi4dprZ4nFAhgaqUlIxhfOVicz8SG33Ol54OEZug=
 github.com/testcontainers/testcontainers-go/modules/vault v0.32.0 h1:ae/zLTcf7is6m2kLAzOJmO5O2Dy/zZV/KZRx6Wx63bw=
 github.com/testcontainers/testcontainers-go/modules/vault v0.32.0/go.mod h1:IOiaemn5bDfDPa8120qECKKKtoOQqKaIKo9pW4bMMes=
 github.com/tetratelabs/wazero v1.8.0 h1:iEKu0d4c2Pd+QSRieYbnQC9yiFlMS9D+Jr0LsRmcF4g=
@@ -1595,12 +1593,6 @@ github.com/tklauser/numcpus v0.8.0 h1:Mx4Wwe/FjZLeQsK/6kt2EOepwwSl7SmJrK5bV/dXYg
 github.com/tklauser/numcpus v0.8.0/go.mod h1:ZJZlAY+dmR4eut8epnzf0u/VwodKmryxR8txiloSqBE=
 github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=
 github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
-github.com/twmb/franz-go v1.16.1 h1:rpWc7fB9jd7TgmCyfxzenBI+QbgS8ZfJOUQE+tzPtbE=
-github.com/twmb/franz-go v1.16.1/go.mod h1:/pER254UPPGp/4WfGqRi+SIRGE50RSQzVubQp6+N4FA=
-github.com/twmb/franz-go/pkg/kadm v1.11.0 h1:FfeWJ0qadntFpAcQt8JzNXW4dijjytZNLrzJuzzzuxA=
-github.com/twmb/franz-go/pkg/kadm v1.11.0/go.mod h1:qrhkdH+SWS3ivmbqOgHbpgVHamhaKcjH0UM+uOp0M1A=
-github.com/twmb/franz-go/pkg/kmsg v1.7.0 h1:a457IbvezYfA5UkiBvyV3zj0Is3y1i8EJgqjJYoij2E=
-github.com/twmb/franz-go/pkg/kmsg v1.7.0/go.mod h1:se9Mjdt0Nwzc9lnjJ0HyDtLyBnaBDAd7pCje47OhSyw=
 github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE=
 github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg=
 github.com/urfave/cli/v2 v2.27.4 h1:o1owoI+02Eb+K107p27wEX9Bb8eqIoZCfLXloLUSWJ8=

_integration-tests/tests/ibm_sarama/ibm_sarama.go

@@ -19,8 +19,7 @@ import (
 	"github.com/IBM/sarama"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/testcontainers/testcontainers-go"
-	"github.com/testcontainers/testcontainers-go/modules/redpanda"
+	"github.com/testcontainers/testcontainers-go/modules/kafka"
 )
 
 const (
@@ -29,31 +28,18 @@ const (
 )
 
 type TestCase struct {
-	server *redpanda.Container
+	server *kafka.KafkaContainer
 	cfg    *sarama.Config
 	addrs  []string
 }
 
 func (tc *TestCase) Setup(t *testing.T) {
-	var (
-		err error
-		ctx = context.Background()
-	)
 	tc.cfg = sarama.NewConfig()
 	tc.cfg.Version = sarama.V0_11_0_0
 	tc.cfg.Producer.Return.Successes = true
 
-	tc.server, err = redpanda.Run(ctx,
-		"docker.redpanda.com/redpandadata/redpanda:v24.2.1",
-		redpanda.WithAutoCreateTopics(),
-		testcontainers.WithLogger(testcontainers.TestLogger(t)),
-		utils.WithTestLogConsumer(t),
-	)
-	utils.AssertTestContainersError(t, err)
-
-	addr, err := tc.server.KafkaSeedBroker(ctx)
-	require.NoError(t, err, "failed to get seed broker address")
-
+	container, addr := utils.StartKafkaTestContainer(t)
+	tc.server = container
 	tc.addrs = []string{addr}
 }
 

_integration-tests/tests/shopify_sarama/shopify_sarama.go

@@ -9,17 +9,16 @@ package shopify_sarama
 
 import (
 	"context"
+	"orchestrion/integration/utils"
 	"testing"
 	"time"
 
-	"orchestrion/integration/utils"
 	"orchestrion/integration/validator/trace"
 
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/testcontainers/testcontainers-go"
-	"github.com/testcontainers/testcontainers-go/modules/redpanda"
+	"github.com/testcontainers/testcontainers-go/modules/kafka"
 )
 
 const (
@@ -28,31 +27,18 @@ const (
 )
 
 type TestCase struct {
-	server *redpanda.Container
+	server *kafka.KafkaContainer
 	cfg    *sarama.Config
 	addrs  []string
 }
 
 func (tc *TestCase) Setup(t *testing.T) {
-	var (
-		err error
-		ctx = context.Background()
-	)
 	tc.cfg = sarama.NewConfig()
 	tc.cfg.Version = sarama.V0_11_0_0
 	tc.cfg.Producer.Return.Successes = true
 
-	tc.server, err = redpanda.Run(ctx,
-		"docker.redpanda.com/redpandadata/redpanda:v24.2.1",
-		redpanda.WithAutoCreateTopics(),
-		testcontainers.WithLogger(testcontainers.TestLogger(t)),
-		utils.WithTestLogConsumer(t),
-	)
-	utils.AssertTestContainersError(t, err)
-
-	addr, err := tc.server.KafkaSeedBroker(ctx)
-	require.NoError(t, err, "failed to get seed broker address")
-
+	container, addr := utils.StartKafkaTestContainer(t)
+	tc.server = container
 	tc.addrs = []string{addr}
 }
 

_integration-tests/utils/testcontainers.go

@@ -7,13 +7,15 @@ package utils
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"runtime"
 	"testing"
 
 	"github.com/docker/go-connections/nat"
 	"github.com/stretchr/testify/require"
 	"github.com/testcontainers/testcontainers-go"
+	"github.com/testcontainers/testcontainers-go/modules/kafka"
 	"github.com/testcontainers/testcontainers-go/wait"
 )
 
@@ -53,6 +55,26 @@ func StartDynamoDBTestContainer(t *testing.T) (testcontainers.Container, string,
 	return server, host, mappedPort.Port()
 }
 
+func StartKafkaTestContainer(t *testing.T) (*kafka.KafkaContainer, string) {
+	ctx := context.Background()
+	exposedPort := "9093/tcp"
+
+	container, err := kafka.Run(ctx,
+		"confluentinc/confluent-local:7.5.0",
+		kafka.WithClusterID("test-cluster"),
+	)
+	AssertTestContainersError(t, err)
+
+	mappedPort, err := container.MappedPort(ctx, nat.Port(exposedPort))
+	require.NoError(t, err)
+
+	host, err := container.Host(ctx)
+	require.NoError(t, err)
+
+	addr := fmt.Sprintf("%s:%s", host, mappedPort.Port())
+	return container, addr
+}
+
 // AssertTestContainersError decides whether the provided testcontainers error should make the test fail or mark it as
 // skipped, depending on the environment where the test is running.
 func AssertTestContainersError(t *testing.T, err error) {