Use KafkaContainer in smoke test for Kafka with SSL

See gh-38359

spring-boot-project/spring-boot-dependencies/build.gradle

@@ -1632,7 +1632,7 @@ bom {
 			]
 		}
 	}
-	library("Testcontainers", "1.19.1") {
+	library("Testcontainers", "1.19.2") {
 		group("org.testcontainers") {
 			imports = [
 				"testcontainers-bom"

spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/build.gradle

@@ -15,4 +15,5 @@ dependencies {
 		exclude group: "commons-logging", module: "commons-logging"
 	}
 	testImplementation("org.testcontainers:junit-jupiter")
+	testImplementation("org.testcontainers:kafka")
 }

spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/test/java/smoketest/kafka/ssl/SampleKafkaSslApplicationTests.java

@@ -16,21 +16,23 @@
 
 package smoketest.kafka.ssl;
 
-import java.io.File;
 import java.time.Duration;
 
 import org.awaitility.Awaitility;
 import org.junit.jupiter.api.Test;
-import org.testcontainers.containers.DockerComposeContainer;
-import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.containers.KafkaContainer;
 import org.testcontainers.junit.jupiter.Container;
 import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.MountableFile;
 import smoketest.kafka.Consumer;
 import smoketest.kafka.Producer;
 import smoketest.kafka.SampleMessage;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.hamcrest.Matchers.empty;
@@ -38,23 +40,37 @@ import static org.hamcrest.Matchers.not;
 
 @Testcontainers(disabledWithoutDocker = true)
 @SpringBootTest(classes = { SampleKafkaSslApplication.class, Producer.class, Consumer.class },
-		properties = { "spring.kafka.security.protocol=SSL", "spring.kafka.bootstrap-servers=localhost:9093",
-				"spring.kafka.ssl.bundle=client",
+		properties = { "spring.kafka.security.protocol=SSL",
+				"spring.kafka.properties.ssl.endpoint.identification.algorithm=", "spring.kafka.ssl.bundle=client",
 				"spring.ssl.bundle.jks.client.keystore.location=classpath:ssl/test-client.p12",
 				"spring.ssl.bundle.jks.client.keystore.password=password",
 				"spring.ssl.bundle.jks.client.truststore.location=classpath:ssl/test-ca.p12",
 				"spring.ssl.bundle.jks.client.truststore.password=password" })
 class SampleKafkaSslApplicationTests {
 
-	private static final File KAFKA_COMPOSE_FILE = new File("src/test/resources/docker-compose.yml");
-
-	private static final String KAFKA_COMPOSE_SERVICE = "kafka";
-
-	private static final int KAFKA_SSL_PORT = 9093;
-
 	@Container
-	public DockerComposeContainer<?> container = new DockerComposeContainer<>(KAFKA_COMPOSE_FILE)
-		.withExposedService(KAFKA_COMPOSE_SERVICE, KAFKA_SSL_PORT, Wait.forListeningPorts(KAFKA_SSL_PORT));
+	public static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.4.0"))
+		.withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "PLAINTEXT:SSL,BROKER:PLAINTEXT")
+		.withEnv("KAFKA_AUTO_CREATE_TOPICS_ENABLE", "true")
+		.withEnv("KAFKA_SSL_CLIENT_AUTH", "required")
+		.withEnv("KAFKA_SSL_KEYSTORE_LOCATION", "/etc/kafka/secrets/certs/test-server.p12")
+		.withEnv("KAFKA_SSL_KEYSTORE_PASSWORD", "password")
+		.withEnv("KAFKA_SSL_KEY_PASSWORD", "password")
+		.withEnv("KAFKA_SSL_TRUSTSTORE_LOCATION", "/etc/kafka/secrets/certs/test-ca.p12")
+		.withEnv("KAFKA_SSL_TRUSTSTORE_PASSWORD", "password")
+		.withEnv("KAFKA_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM", "")
+		.withCopyFileToContainer(MountableFile.forClasspathResource("ssl/test-server.p12"),
+				"/etc/kafka/secrets/certs/test-server.p12")
+		.withCopyFileToContainer(MountableFile.forClasspathResource("ssl/credentials"),
+				"/etc/kafka/secrets/certs/credentials")
+		.withCopyFileToContainer(MountableFile.forClasspathResource("ssl/test-ca.p12"),
+				"/etc/kafka/secrets/certs/test-ca.p12");
+
+	@DynamicPropertySource
+	static void kafkaProperties(DynamicPropertyRegistry registry) {
+		registry.add("spring.kafka.bootstrap-servers",
+				() -> String.format("%s:%s", kafka.getHost(), kafka.getMappedPort(9093)));
+	}
 
 	@Autowired
 	private Producer producer;

spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-kafka/src/test/resources/docker-compose.yml

@@ -1,30 +0,0 @@
----
-version: '2'
-services:
-  zookeeper:
-    image: confluentinc/cp-zookeeper:7.4.0
-    environment:
-      ZOOKEEPER_CLIENT_PORT: 2181
-      ZOOKEEPER_TICK_TIME: 2000
-
-  kafka:
-    image: confluentinc/cp-kafka:7.4.0
-    depends_on:
-      - zookeeper
-    ports:
-      - "9092:9092"
-      - "9093:9093"
-    environment:
-      KAFKA_BROKER_ID: 1
-      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,SSL://localhost:9093
-      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
-      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
-      KAFKA_SSL_CLIENT_AUTH: "required"
-      KAFKA_SSL_KEYSTORE_FILENAME: '/certs/test-server.p12'
-      KAFKA_SSL_KEYSTORE_CREDENTIALS: '/certs/credentials'
-      KAFKA_SSL_KEY_CREDENTIALS: '/certs/credentials'
-      KAFKA_SSL_TRUSTSTORE_FILENAME: '/certs/test-ca.p12'
-      KAFKA_SSL_TRUSTSTORE_CREDENTIALS: '/certs/credentials'
-    volumes:
-      - ./ssl/:/etc/kafka/secrets/certs