fix(test): resolve flaky Kafka Connect integration test

- Fix cleanup order: terminate containers before removing network
- Replace log-based wait with HTTP health check for reliability
- Pin connectors image to digest for reproducibility
- Add log consumer for debugging container failures
- Upgrade Redpanda version to v25.2.1
- Remove WithAttachable flag for consistency

Author: weeco

backend/pkg/api/handle_kafka_connect_integration_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/docker/go-connections/nat"
 	"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/network"
 
@@ -42,14 +43,11 @@ func (s *APIIntegrationTestSuite) TestHandleCreateConnector() {
 	ctx := t.Context()
 
 	// create one common network that all containers will share
-	testNetwork, err := network.New(ctx, network.WithAttachable())
+	testNetwork, err := network.New(ctx)
 	require.NoError(err)
-	t.Cleanup(func() {
-		assert.NoError(testNetwork.Remove(context.Background()))
-	})
 
 	redpandaContainer, err := redpanda.Run(ctx,
-		"redpandadata/redpanda:v23.3.18",
+		"redpandadata/redpanda:v25.2.1",
 		network.WithNetwork([]string{"redpanda"}, testNetwork),
 		redpanda.WithListener("redpanda:29092"),
 	)
@@ -58,15 +56,29 @@ func (s *APIIntegrationTestSuite) TestHandleCreateConnector() {
 	seedBroker, err := redpandaContainer.KafkaSeedBroker(ctx)
 	require.NoError(err)
 
+	// Log consumer for debugging container failures
+	logConsumer := &testutil.TestContainersLogger{
+		LogPrefix: "[kafka-connect] ",
+	}
+
 	// Kafka KafkaConnect container
-	connectC, err := testutil.RunRedpandaConnectorsContainer(
+	connectContainer, err := testutil.RunRedpandaConnectorsContainer(
 		ctx,
 		[]string{"redpanda:29092"},
 		network.WithNetwork([]string{"kafka-connect"}, testNetwork),
+		testcontainers.WithLogConsumers(logConsumer),
 	)
 	require.NoError(err)
 
-	connectContainer := connectC
+	// Register cleanup in correct order: containers first, then network
+	t.Cleanup(func() {
+		cleanupCtx := context.Background()
+		// Terminate containers first (LIFO: registered last, runs first)
+		_ = connectContainer.Terminate(cleanupCtx)
+		_ = redpandaContainer.Terminate(cleanupCtx)
+		// Then remove network
+		_ = testNetwork.Remove(cleanupCtx)
+	})
 
 	connectPort, err := connectContainer.MappedPort(ctx, nat.Port("8083"))
 	require.NoError(err)
@@ -104,11 +116,6 @@ func (s *APIIntegrationTestSuite) TestHandleCreateConnector() {
 		s.api.ConnectSvc = oldConnectSvc
 	}()
 
-	t.Cleanup(func() {
-		assert.NoError(connectContainer.Terminate(context.Background()))
-		assert.NoError(redpandaContainer.Terminate(context.Background()))
-	})
-
 	t.Run("happy path", func(t *testing.T) {
 		ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
 		defer cancel()

backend/pkg/testutil/kafkaconnect.go

@@ -38,7 +38,8 @@ func RunRedpandaConnectorsContainer(ctx context.Context, bootstrapServers []stri
 
 	request := testcontainers.GenericContainerRequest{
 		ContainerRequest: testcontainers.ContainerRequest{
-			Image:        "docker.cloudsmith.io/redpanda/connectors-unsupported/connectors:latest",
+			// Pin to digest for reproducibility - :latest tag can have variable behavior
+			Image:        "docker.cloudsmith.io/redpanda/connectors-unsupported/connectors@sha256:0ff21e793ef3042f2f48fb3d6549fae0ef687950a76b8017ab9d8b17c33cadb0",
 			ExposedPorts: []string{"8083/tcp"},
 			Env: map[string]string{
 				"CONNECT_CONFIGURATION":     testConnectConfig,
@@ -47,11 +48,15 @@ func RunRedpandaConnectorsContainer(ctx context.Context, bootstrapServers []stri
 				"CONNECT_HEAP_OPTS":         "-Xms512M -Xmx512M",
 				"CONNECT_LOG_LEVEL":         "info",
 			},
-			WaitingFor: wait.ForAll(
-				wait.ForLog("Kafka Connect started").
-					WithPollInterval(500 * time.Millisecond).
-					WithStartupTimeout(waitTimeout),
-			),
+			// Use HTTP health check instead of log-based wait for better reliability
+			// and early failure detection if the container exits
+			WaitingFor: wait.ForHTTP("/connectors").
+				WithPort("8083/tcp").
+				WithStatusCodeMatcher(func(status int) bool {
+					return status == 200
+				}).
+				WithPollInterval(1 * time.Second).
+				WithStartupTimeout(waitTimeout),
 		},
 		Started: true,
 	}

frontend/tests/shared/global-setup.mjs

@@ -240,7 +240,8 @@ topic.creation.enable=false
 
   let connect;
   try {
-    connect = await new GenericContainer('docker.cloudsmith.io/redpanda/connectors-unsupported/connectors:latest')
+    // Pin to digest for reproducibility - :latest tag can have variable behavior
+    connect = await new GenericContainer('docker.cloudsmith.io/redpanda/connectors-unsupported/connectors@sha256:0ff21e793ef3042f2f48fb3d6549fae0ef687950a76b8017ab9d8b17c33cadb0')
       .withPlatform('linux/amd64')
       .withNetwork(network)
       .withNetworkAliases('connect')