Skip to content

Commit 3ae216a

Browse files
PerieclePeriecle
authored
Embedded Native Kafka introduced (#2444)
1 parent 5f4477a commit 3ae216a

23 files changed

Lines changed: 1642 additions & 0 deletions

‎README.adoc‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -267,6 +267,8 @@ embedded:
267267

268268
=== link:embedded-kafka/README.adoc[embedded-kafka]
269269

270+
=== link:embedded-native-kafka/README.adoc[embedded-native-kafka]
271+
270272
=== link:embedded-rabbitmq/README.adoc[embedded-rabbitmq]
271273

272274
=== link:embedded-aerospike/README.adoc[embedded-aerospike]
@@ -377,3 +379,4 @@ embedded:
377379
** mailto:[email protected][[email protected]]
378380
** mailto:[email protected][[email protected]]
379381
** mailto:[email protected][[email protected]]
382+
** mailto:[email protected][[email protected]]

‎SECURITY.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,3 +16,4 @@ If you have the vulnerability report. Please contact with :
1616
- [[email protected]](mailto:[email protected])
1717
- [[email protected]](mailto:[email protected])
1818
- [[email protected]](mailto:[email protected])
19+
- [[email protected]](mailto:[email protected])

‎embedded-native-kafka/README.adoc‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
1+
=== embedded-native-kafka
2+
3+
This module provides Apache Kafka Native (GraalVM) support for integration testing using TestContainers.
4+
The native Kafka image uses KRaft instead of Zookeeper, providing a simpler and more efficient setup.
5+
6+
==== Maven dependency
7+
8+
.pom.xml
9+
[source,xml]
10+
----
11+
<dependency>
12+
<groupId>com.playtika.testcontainers</groupId>
13+
<artifactId>embedded-native-kafka</artifactId>
14+
<scope>test</scope>
15+
</dependency>
16+
----
17+
18+
==== Consumes (via `bootstrap.properties`)
19+
* `embedded.nativekafka.enabled` `(true|false, default is true)`
20+
* `embedded.nativekafka.topicsToCreate` `(comma separated list of topic names, default is empty)`
21+
* `embedded.nativekafka.dockerImage` `(default is set to 'apache/kafka-native:4.0.0')`
22+
** Image versions on https://hub.docker.com/r/apache/kafka-native/tags[dockerhub]
23+
* `embedded.nativekafka.waitTimeoutInSeconds` `(default is 60 seconds)`
24+
* `embedded.nativekafka.kafkaPort` `(default is 9092)`
25+
26+
==== Filesystem bindings
27+
28+
Container for `embedded-native-kafka` can bind its volumes to host filesystem.
29+
By default, to your projects `target` folder. You can configure binding using properties:
30+
31+
* `embedded.nativekafka.fileSystemBind.enabled` `(true|false, default is false)`
32+
* `embedded.nativekafka.fileSystemBind.dataFolder` `(default : target/embedded-native-kafka-data)`
33+
34+
==== Produces
35+
36+
* `embedded.nativekafka.bootstrapServers`
37+
* `embedded.nativekafka.brokerList`
38+
* `embedded.nativekafka.networkAlias`
39+
* `embedded.nativekafka.host`
40+
* `embedded.nativekafka.port`
41+
42+
==== Example configuration
43+
44+
.`embedded-native-kafka` configuration (via `bootstrap.properties`)
45+
[source,properties]
46+
----
47+
embedded.nativekafka.topicsToCreate=topic1,topic2
48+
embedded.nativekafka.fileSystemBind.enabled=true
49+
----
50+
51+
.Application configuration
52+
[source,properties]
53+
----
54+
# Kafka configuration
55+
bootstrap.servers=${embedded.nativekafka.bootstrapServers}
56+
57+
# Consumer configuration
58+
spring.kafka.consumer.bootstrap-servers=${embedded.nativekafka.bootstrapServers}
59+
spring.kafka.consumer.group-id=test-group
60+
61+
# Producer configuration
62+
spring.kafka.producer.bootstrap-servers=${embedded.nativekafka.bootstrapServers}
63+
----
64+
65+
==== Java usage example
66+
67+
[source,java]
68+
----
69+
@SpringBootTest
70+
class EmbeddedNativeKafkaTest {
71+
72+
@Autowired
73+
private KafkaTemplate<String, String> kafkaTemplate;
74+
75+
@Value("${embedded.nativekafka.bootstrapServers}")
76+
private String bootstrapServers;
77+
78+
@Test
79+
void testKafkaMessageSendAndReceive() {
80+
// Your test logic here
81+
kafkaTemplate.send("test-topic", "test-message");
82+
83+
// Assert message consumption
84+
// ...
85+
}
86+
}

‎embedded-native-kafka/pom.xml‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<project xmlns="http://maven.apache.org/POM/4.0.0"
3+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5+
<modelVersion>4.0.0</modelVersion>
6+
7+
<parent>
8+
<artifactId>testcontainers-spring-boot-parent</artifactId>
9+
<groupId>com.playtika.testcontainers</groupId>
10+
<version>3.1.14</version>
11+
<relativePath>../testcontainers-spring-boot-parent</relativePath>
12+
</parent>
13+
14+
<artifactId>embedded-native-kafka</artifactId>
15+
16+
<properties>
17+
<kafka-clients.version>4.0.0</kafka-clients.version>
18+
<maven-clean-plugin.version>3.5.0</maven-clean-plugin.version>
19+
</properties>
20+
21+
<dependencyManagement>
22+
<dependencies>
23+
<dependency>
24+
<groupId>org.apache.kafka</groupId>
25+
<artifactId>kafka-clients</artifactId>
26+
<version>${kafka-clients.version}</version>
27+
</dependency>
28+
</dependencies>
29+
</dependencyManagement>
30+
31+
<dependencies>
32+
<dependency>
33+
<groupId>com.playtika.testcontainers</groupId>
34+
<artifactId>testcontainers-common</artifactId>
35+
</dependency>
36+
<dependency>
37+
<groupId>org.testcontainers</groupId>
38+
<artifactId>kafka</artifactId>
39+
</dependency>
40+
<dependency>
41+
<groupId>jakarta.validation</groupId>
42+
<artifactId>jakarta.validation-api</artifactId>
43+
</dependency>
44+
<dependency>
45+
<groupId>org.apache.kafka</groupId>
46+
<artifactId>kafka-clients</artifactId>
47+
</dependency>
48+
<dependency>
49+
<groupId>org.springframework.boot</groupId>
50+
<artifactId>spring-boot-starter-web</artifactId>
51+
<scope>test</scope>
52+
<exclusions>
53+
<exclusion>
54+
<groupId>org.springframework.boot</groupId>
55+
<artifactId>spring-boot-starter-logging</artifactId>
56+
</exclusion>
57+
</exclusions>
58+
</dependency>
59+
</dependencies>
60+
61+
<build>
62+
<plugins>
63+
<plugin>
64+
<groupId>org.apache.maven.plugins</groupId>
65+
<artifactId>maven-clean-plugin</artifactId>
66+
<version>${maven-clean-plugin.version}</version>
67+
<executions>
68+
<execution>
69+
<id>prepare-tests</id>
70+
<!-- Delete eventual artifacts from previous test mounts -->
71+
<phase>test-compile</phase>
72+
<goals>
73+
<goal>clean</goal>
74+
</goals>
75+
</execution>
76+
</executions>
77+
<configuration>
78+
<failOnError>false</failOnError>
79+
<excludeDefaultDirectories>true</excludeDefaultDirectories>
80+
<filesets>
81+
<fileset>
82+
<directory>${java.io.tmpdir}</directory>
83+
<followSymlinks>false</followSymlinks>
84+
<useDefaultExcludes>true</useDefaultExcludes>
85+
<includes>
86+
<include>embedded-native-kafka-*/**</include>
87+
</includes>
88+
</fileset>
89+
</filesets>
90+
</configuration>
91+
</plugin>
92+
</plugins>
93+
</build>
94+
</project>

‎embedded-native-kafka/src/main/java/com/playtika/testcontainer/nativekafka/NativeKafkaTopicsConfigurer.java‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
package com.playtika.testcontainer.nativekafka;
2+
3+
import com.playtika.testcontainer.nativekafka.properties.NativeKafkaConfigurationProperties;
4+
import com.playtika.testcontainer.nativekafka.properties.NativeKafkaConfigurationProperties.TopicConfiguration;
5+
import lombok.Getter;
6+
import lombok.RequiredArgsConstructor;
7+
import lombok.extern.slf4j.Slf4j;
8+
import org.apache.kafka.clients.admin.AdminClient;
9+
import org.apache.kafka.clients.admin.CreateTopicsResult;
10+
import org.apache.kafka.clients.admin.NewTopic;
11+
import org.apache.kafka.clients.producer.ProducerConfig;
12+
import org.springframework.beans.factory.InitializingBean;
13+
import org.testcontainers.containers.GenericContainer;
14+
import org.testcontainers.kafka.KafkaContainer;
15+
16+
import java.util.Collection;
17+
import java.util.List;
18+
import java.util.Map;
19+
import java.util.Properties;
20+
import java.util.concurrent.ExecutionException;
21+
import java.util.stream.Collectors;
22+
23+
import static java.util.stream.Collectors.toMap;
24+
25+
@Slf4j
26+
@RequiredArgsConstructor
27+
@Getter
28+
public class NativeKafkaTopicsConfigurer implements InitializingBean {
29+
private static final int DEFAULT_PARTITION_COUNT = 1;
30+
31+
private final GenericContainer<?> nativeKafka;
32+
private final NativeKafkaConfigurationProperties nativeKafkaProperties;
33+
34+
@Override
35+
public void afterPropertiesSet() {
36+
createTopics(this.nativeKafkaProperties.getTopicsToCreate(), this.nativeKafkaProperties.getTopicsConfiguration());
37+
}
38+
39+
public void createTopics(Collection<String> topics, Collection<TopicConfiguration> topicsConfiguration) {
40+
Map<String, TopicConfiguration> defaultTopicToTopicConfigurationMap =
41+
topics.stream()
42+
.collect(toMap(topic -> topic,
43+
topic -> new TopicConfiguration(topic, DEFAULT_PARTITION_COUNT)));
44+
45+
Map<String, TopicConfiguration> topicToTopicConfigurationMap =
46+
topicsConfiguration.stream()
47+
.collect(toMap(TopicConfiguration::getTopic,
48+
topicConfiguration -> topicConfiguration));
49+
50+
defaultTopicToTopicConfigurationMap.putAll(topicToTopicConfigurationMap);
51+
52+
Collection<TopicConfiguration> topicsConfigurationToCreate = defaultTopicToTopicConfigurationMap.values();
53+
54+
if (!topicsConfigurationToCreate.isEmpty()) {
55+
log.info("Creating Native Kafka topics for configuration: {}", topicsConfigurationToCreate);
56+
createTopicsUsingAdminClient(topicsConfigurationToCreate);
57+
log.info("Created Native Kafka topics for configuration: {}", topicsConfigurationToCreate);
58+
}
59+
}
60+
61+
private void createTopicsUsingAdminClient(Collection<TopicConfiguration> topicsConfigurationToCreate) {
62+
Properties adminProps = new Properties();
63+
adminProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ((KafkaContainer) nativeKafka).getBootstrapServers());
64+
65+
try (AdminClient adminClient = AdminClient.create(adminProps)) {
66+
List<NewTopic> newTopics = topicsConfigurationToCreate.stream()
67+
.map(config -> new NewTopic(config.getTopic(), config.getPartitions(), (short) 1))
68+
.collect(Collectors.toList());
69+
70+
CreateTopicsResult result = adminClient.createTopics(newTopics);
71+
72+
// Wait for all topics to be created
73+
result.all().get();
74+
log.debug("Successfully created {} topics using Admin API", newTopics.size());
75+
} catch (InterruptedException | ExecutionException e) {
76+
log.error("Failed to create topics using Admin API", e);
77+
throw new RuntimeException("Failed to create topics", e);
78+
}
79+
}
80+
}

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

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
package com.playtika.testcontainer.nativekafka.configuration;
2+
3+
import com.playtika.testcontainer.common.spring.DockerPresenceBootstrapConfiguration;
4+
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
5+
import org.springframework.boot.autoconfigure.AutoConfigureOrder;
6+
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
7+
import org.springframework.context.annotation.Configuration;
8+
import org.springframework.context.annotation.Import;
9+
10+
@Configuration
11+
@AutoConfigureOrder
12+
@ConditionalOnExpression("${embedded.containers.enabled:true}")
13+
@AutoConfigureAfter(DockerPresenceBootstrapConfiguration.class)
14+
public class EmbeddedNativeKafkaBootstrapConfiguration {
15+
16+
@Import(value = {
17+
NativeKafkaContainerConfiguration.class,
18+
})
19+
@Configuration
20+
public static class AllConfigurations {
21+
}
22+
}

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

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
package com.playtika.testcontainer.nativekafka.configuration;
2+
3+
import com.playtika.testcontainer.common.properties.InstallPackageProperties;
4+
import com.playtika.testcontainer.common.utils.PackageInstaller;
5+
import com.playtika.testcontainer.common.utils.YumPackageInstaller;
6+
import com.playtika.testcontainer.nativekafka.properties.NativeKafkaConfigurationProperties;
7+
import org.springframework.beans.factory.annotation.Qualifier;
8+
import org.springframework.boot.autoconfigure.AutoConfiguration;
9+
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
10+
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
11+
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
12+
import org.springframework.boot.context.properties.ConfigurationProperties;
13+
import org.springframework.context.annotation.Bean;
14+
import org.testcontainers.containers.GenericContainer;
15+
16+
import static com.playtika.testcontainer.nativekafka.properties.NativeKafkaConfigurationProperties.NATIVE_KAFKA_BEAN_NAME;
17+
import static com.playtika.testcontainer.nativekafka.properties.NativeKafkaConfigurationProperties.NATIVE_KAFKA_PACKAGE_PROPERTIES_BEAN_NAME;
18+
19+
@AutoConfiguration
20+
@ConditionalOnBean({NativeKafkaConfigurationProperties.class})
21+
@ConditionalOnExpression("${embedded.containers.enabled:true}")
22+
@ConditionalOnProperty(value = {"embedded.nativekafka.enabled"}, havingValue = "true", matchIfMissing = true)
23+
public class EmbeddedNativeKafkaTestOperationsAutoConfiguration {
24+
25+
@Bean(NATIVE_KAFKA_PACKAGE_PROPERTIES_BEAN_NAME)
26+
@ConfigurationProperties("embedded.nativekafka.install")
27+
public InstallPackageProperties nativeKafkaPackageProperties() {
28+
return new InstallPackageProperties();
29+
}
30+
31+
@Bean
32+
public PackageInstaller nativeKafkaPackageInstaller(@Qualifier(NATIVE_KAFKA_PACKAGE_PROPERTIES_BEAN_NAME) InstallPackageProperties nativeKafkaPackageProperties,
33+
@Qualifier(NATIVE_KAFKA_BEAN_NAME) GenericContainer<?> nativeKafka) {
34+
return new YumPackageInstaller(nativeKafkaPackageProperties, nativeKafka);
35+
}
36+
}

0 commit comments

Comments
 (0)