fix: Use same properties for Native Kafka that are used for Confluent Kafka (#2482)

Author: Periecle

embedded-native-kafka/README.adoc

@@ -3,6 +3,8 @@
 This module provides Apache Kafka Native (GraalVM) support for integration testing using TestContainers.
 The native Kafka image uses KRaft instead of Zookeeper, providing a simpler and more efficient setup.
 
+WARNING: Do not try to use both embedded-kafka and embedded-native-kafka!
+
 ==== Maven dependency
 
 .pom.xml
@@ -16,50 +18,50 @@ The native Kafka image uses KRaft instead of Zookeeper, providing a simpler and
 ----
 
 ==== Consumes (via `bootstrap.properties`)
-* `embedded.nativekafka.enabled` `(true|false, default is true)`
-* `embedded.nativekafka.topicsToCreate` `(comma separated list of topic names, default is empty)`
-* `embedded.nativekafka.dockerImage` `(default is set to 'apache/kafka-native:4.0.0')`
+* `embedded.kafka.enabled` `(true|false, default is true)`
+* `embedded.kafka.topicsToCreate` `(comma separated list of topic names, default is empty)`
+* `embedded.kafka.dockerImage` `(default is set to 'apache/kafka-native:4.0.0')`
 ** Image versions on https://hub.docker.com/r/apache/kafka-native/tags[dockerhub]
-* `embedded.nativekafka.waitTimeoutInSeconds` `(default is 60 seconds)`
-* `embedded.nativekafka.kafkaPort` `(default is 9092)`
+* `embedded.kafka.waitTimeoutInSeconds` `(default is 60 seconds)`
+* `embedded.kafka.kafkaPort` `(default is 9092)`
 
 ==== Filesystem bindings
 
 Container for `embedded-native-kafka` can bind its volumes to host filesystem.
 By default, to your projects `target` folder. You can configure binding using properties:
 
-* `embedded.nativekafka.fileSystemBind.enabled` `(true|false, default is false)`
-* `embedded.nativekafka.fileSystemBind.dataFolder` `(default : target/embedded-native-kafka-data)`
+* `embedded.kafka.fileSystemBind.enabled` `(true|false, default is false)`
+* `embedded.kafka.fileSystemBind.dataFolder` `(default : target/embedded-native-kafka-data)`
 
 ==== Produces
 
-* `embedded.nativekafka.bootstrapServers`
-* `embedded.nativekafka.brokerList`
-* `embedded.nativekafka.networkAlias`
-* `embedded.nativekafka.host`
-* `embedded.nativekafka.port`
+* `embedded.kafka.bootstrapServers`
+* `embedded.kafka.brokerList`
+* `embedded.kafka.networkAlias`
+* `embedded.kafka.host`
+* `embedded.kafka.port`
 
 ==== Example configuration
 
 .`embedded-native-kafka` configuration (via `bootstrap.properties`)
 [source,properties]
 ----
-embedded.nativekafka.topicsToCreate=topic1,topic2
-embedded.nativekafka.fileSystemBind.enabled=true
+embedded.kafka.topicsToCreate=topic1,topic2
+embedded.kafka.fileSystemBind.enabled=true
 ----
 
 .Application configuration
 [source,properties]
 ----
 # Kafka configuration
-bootstrap.servers=${embedded.nativekafka.bootstrapServers}
+bootstrap.servers=${embedded.kafka.bootstrapServers}
 
 # Consumer configuration
-spring.kafka.consumer.bootstrap-servers=${embedded.nativekafka.bootstrapServers}
+spring.kafka.consumer.bootstrap-servers=${embedded.kafka.bootstrapServers}
 spring.kafka.consumer.group-id=test-group
 
 # Producer configuration  
-spring.kafka.producer.bootstrap-servers=${embedded.nativekafka.bootstrapServers}
+spring.kafka.producer.bootstrap-servers=${embedded.kafka.bootstrapServers}
 ----
 
 ==== Java usage example
@@ -72,7 +74,7 @@ class EmbeddedNativeKafkaTest {
     @Autowired
     private KafkaTemplate<String, String> kafkaTemplate;
 
-    @Value("${embedded.nativekafka.bootstrapServers}")
+    @Value("${embedded.kafka.bootstrapServers}")
     private String bootstrapServers;
 
     @Test

embedded-native-kafka/src/main/java/com/playtika/testcontainer/nativekafka/configuration/EmbeddedNativeKafkaTestOperationsAutoConfiguration.java

@@ -19,11 +19,11 @@
 @AutoConfiguration
 @ConditionalOnBean({NativeKafkaConfigurationProperties.class})
 @ConditionalOnExpression("${embedded.containers.enabled:true}")
-@ConditionalOnProperty(value = {"embedded.nativekafka.enabled"}, havingValue = "true", matchIfMissing = true)
+@ConditionalOnProperty(value = {"embedded.kafka.enabled"}, havingValue = "true", matchIfMissing = true)
 public class EmbeddedNativeKafkaTestOperationsAutoConfiguration {
 
     @Bean(NATIVE_KAFKA_PACKAGE_PROPERTIES_BEAN_NAME)
-    @ConfigurationProperties("embedded.nativekafka.install")
+    @ConfigurationProperties("embedded.kafka.install")
     public InstallPackageProperties nativeKafkaPackageProperties() {
         return new InstallPackageProperties();
     }

embedded-native-kafka/src/main/java/com/playtika/testcontainer/nativekafka/configuration/NativeKafkaContainerConfiguration.java

@@ -35,11 +35,11 @@
 
 @Slf4j
 @Configuration
-@ConditionalOnProperty(value = "embedded.nativekafka.enabled", havingValue = "true", matchIfMissing = true)
+@ConditionalOnProperty(value = "embedded.kafka.enabled", havingValue = "true", matchIfMissing = true)
 @EnableConfigurationProperties(value = {NativeKafkaConfigurationProperties.class})
 public class NativeKafkaContainerConfiguration {
 
-    public static final String NATIVE_KAFKA_HOST_NAME = "native-kafka-broker.testcontainer.docker";
+    public static final String NATIVE_KAFKA_HOST_NAME = "kafka-broker.testcontainer.docker";
 
     @Bean(destroyMethod = "close")
     @ConditionalOnMissingBean(Network.class)
@@ -102,13 +102,13 @@ private void registerNativeKafkaEnvironment(GenericContainer<?> nativeKafka,
         LinkedHashMap<String, Object> map = new LinkedHashMap<>();
 
         String bootstrapServers = ((KafkaContainer) nativeKafka).getBootstrapServers();
-        map.put("embedded.nativekafka.bootstrapServers", bootstrapServers);
-        map.put("embedded.nativekafka.brokerList", bootstrapServers);
-        map.put("embedded.nativekafka.networkAlias", NATIVE_KAFKA_HOST_NAME);
-        map.put("embedded.nativekafka.host", nativeKafka.getHost());
-        map.put("embedded.nativekafka.port", nativeKafka.getMappedPort(nativeKafkaProperties.getKafkaPort()));
+        map.put("embedded.kafka.bootstrapServers", bootstrapServers);
+        map.put("embedded.kafka.brokerList", bootstrapServers);
+        map.put("embedded.kafka.networkAlias", NATIVE_KAFKA_HOST_NAME);
+        map.put("embedded.kafka.host", nativeKafka.getHost());
+        map.put("embedded.kafka.port", nativeKafka.getMappedPort(nativeKafkaProperties.getKafkaPort()));
 
-        MapPropertySource propertySource = new MapPropertySource("embeddedNativeKafkaInfo", map);
+        MapPropertySource propertySource = new MapPropertySource("embeddedKafkaInfo", map);
 
         log.info("Started native kafka broker. Connection details: {}", map);
 

embedded-native-kafka/src/main/java/com/playtika/testcontainer/nativekafka/properties/NativeKafkaConfigurationProperties.java

@@ -9,13 +9,13 @@
 import lombok.With;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 
 @Data
 @EqualsAndHashCode(callSuper = true)
-@ConfigurationProperties("embedded.nativekafka")
+@ConfigurationProperties("embedded.kafka")
 public class NativeKafkaConfigurationProperties extends CommonContainerProperties {
 
     public static final String NATIVE_KAFKA_BEAN_NAME = "nativeKafka";
@@ -32,7 +32,7 @@ public class NativeKafkaConfigurationProperties extends CommonContainerPropertie
     protected FileSystemBind fileSystemBind = new FileSystemBind();
 
     public NativeKafkaConfigurationProperties() {
-        this.setCapabilities(Arrays.asList(Capability.NET_ADMIN));
+        this.setCapabilities(List.of(Capability.NET_ADMIN));
     }
 
     @Override

embedded-native-kafka/src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -3,22 +3,22 @@
   ],
   "properties": [
     {
-      "name": "embedded.nativekafka.enabled",
+      "name": "embedded.kafka.enabled",
       "type": "java.lang.Boolean",
       "defaultValue": "true"
     }
   ],
   "hints": [
     {
-      "name": "embedded.nativekafka.enabled",
+      "name": "embedded.kafka.enabled",
       "values": [
         {
           "value": "true",
-          "description": "Enables configuration of native kafka cluster on startup."
+          "description": "Enables configuration of kafka cluster on startup."
         },
         {
           "value": "false",
-          "description": "Disabled configuration of native kafka cluster on startup."
+          "description": "Disables configuration of kafka cluster on startup."
         }
       ]
     }

embedded-native-kafka/src/test/java/com/playtika/testcontainer/nativekafka/AbstractEmbeddedNativeKafkaTest.java

@@ -56,17 +56,17 @@
 
 @SpringBootTest(
         properties = {
-                "embedded.nativekafka.topicsToCreate=topic1,topic2,topic3",
-                "embedded.nativekafka.topicsConfiguration[0].topic=topic3",
-                "embedded.nativekafka.topicsConfiguration[0].partitions=2"
+                "embedded.kafka.topicsToCreate=topic1,topic2,topic3",
+                "embedded.kafka.topicsConfiguration[0].topic=topic3",
+                "embedded.kafka.topicsConfiguration[0].partitions=2"
         },
         classes = {AbstractEmbeddedNativeKafkaTest.TestConfiguration.class}
 )
 public abstract class AbstractEmbeddedNativeKafkaTest {
 
     @Autowired
     protected AdminClient adminClient;
-    @Value("${embedded.nativekafka.bootstrapServers}")
+    @Value("${embedded.kafka.bootstrapServers}")
     protected List<String> nativeKafkaBrokerList;
 
     protected void sendMessage(String topic, String message) throws Exception {
@@ -272,7 +272,7 @@ protected static Path projectDir() throws Exception {
     static class TestConfiguration {
 
         @Bean
-        public AdminClient adminClient(@Value("${embedded.nativekafka.bootstrapServers}") String nativeKafkaBrokerList) {
+        public AdminClient adminClient(@Value("${embedded.kafka.bootstrapServers}") String nativeKafkaBrokerList) {
             Properties config = new Properties();
             config.put(BOOTSTRAP_SERVERS_CONFIG, nativeKafkaBrokerList);
             return AdminClient.create(config);

embedded-native-kafka/src/test/java/com/playtika/testcontainer/nativekafka/DisableNativeKafkaTest.java

@@ -23,7 +23,7 @@ public class DisableNativeKafkaTest {
     public void contextLoads() {
         contextRunner
                 .withPropertyValues(
-                        "embedded.nativekafka.enabled=false"
+                        "embedded.kafka.enabled=false"
                 )
                 .run((context) -> assertThat(context)
                         .hasNotFailed()

embedded-native-kafka/src/test/java/com/playtika/testcontainer/nativekafka/EmbeddedNativeKafkaTest.java

@@ -18,7 +18,7 @@
 
 @Order(2)
 @TestPropertySource(properties = {
-    "embedded.nativekafka.fileSystemBind.dataFolder=${java.io.tmpdir}/embedded-native-kafka-data-unexpected"
+    "embedded.kafka.fileSystemBind.dataFolder=${java.io.tmpdir}/embedded-native-kafka-data-unexpected"
 })
 @TestInstance(PER_CLASS)
 @DisplayName("Default embedded-native-kafka setup test")

embedded-native-kafka/src/test/java/com/playtika/testcontainer/nativekafka/EmbeddedNativeKafkaWithBindingTest.java

@@ -21,9 +21,9 @@
 
 @Order(6)
 @TestPropertySource(properties = {
-    "embedded.nativekafka.waitTimeoutInSeconds=120",
-    "embedded.nativekafka.fileSystemBind.enabled=true",
-    "embedded.nativekafka.fileSystemBind.dataFolder=${java.io.tmpdir}/embedded-native-kafka-data" // /tmp is more permissible than target
+    "embedded.kafka.waitTimeoutInSeconds=120",
+    "embedded.kafka.fileSystemBind.enabled=true",
+    "embedded.kafka.fileSystemBind.dataFolder=${java.io.tmpdir}/embedded-native-kafka-data" // /tmp is more permissible than target
 })
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 @DisplayName("Test that embedded-native-kafka with filesystem binding")

embedded-native-kafka/src/test/java/com/playtika/testcontainer/nativekafka/configuration/NativeKafkaContainerConfigurationTest.java

@@ -252,7 +252,7 @@ void shouldHandleParentDirectoryCreation() throws IOException {
     @DisplayName("should verify container host name constant")
     void shouldVerifyContainerHostNameConstant() {
         assertThat(NativeKafkaContainerConfiguration.NATIVE_KAFKA_HOST_NAME)
-                .isEqualTo("native-kafka-broker.testcontainer.docker");
+                .isEqualTo("kafka-broker.testcontainer.docker");
     }
 
     @Test