[Enhancement]: Kafka Module: graceful shutdown

[JordanP]

Proposal

Currently, the Kafka module spawns a container that doesn't properly handle signals (term/int). The following (next) unit test reproduces the issue:

func TestKafkaGracefulShutdown(t *testing.T) {
	ctx := context.Background()
	kafkaContainer, err := RunContainer(ctx, WithClusterID("kraftCluster"), testcontainers.WithImage("confluentinc/confluent-local:7.5.0"))
	if err != nil {
		t.Fatal(err)
	}

	done := make(chan struct{})
	go func() {
		// Wait 60 seconds, after that the pod will be forcefully terminated.
		timeout := 60 * time.Second
		kafkaContainer.Stop(context.Background(), &timeout)
		done <- struct{}{}
	}()
	select {
	case <-done:
	case <-time.After(50 * time.Second): // 50 secs should be plenty for Kafka to properly handle the signal
		t.Fatalf("Kafka did not gracefully exit in due time")
	}
}

I tried to replace /etc/confluent/docker/launch with exec /etc/confluent/docker/launch and starterScript with exec starterScript but that didn't do the trick.

[JordanP]

Interestingly, using confluentinc/cp-kafka:7.5.3 and exec /etc/confluent/docker/launch and exec " + starterScript work just fine.

[mdelapenya]

Summoning @eddumelendez as our Kafka expert 🙏

[eddumelendez]

It could be a difference between both images cp-kafka and confluent-local images. You can raise an issue here. I also found this old issue but it is just for cp-kafka

[JordanP]

Do you remember/know why this Kafka module uses the confluent-local variant and not the (more widely used / less experimental ?) cp-kafka ?

[eddumelendez]

I think the same implementation will work with confluentinc/cp-kafka but only in zookeeperless (raft) mode. IIRC, in go we decided to start with confluent-local as it is declared to be used for development workflows.