Skip to content

Commit 769f3e9

Browse files
eddumelendezwilkinsona
authored andcommitted
Add support for org.testcontainers.kafka.KafkaContainer
Testcontainers 1.19.8 provides `org.testcontainers.kafka.KafkaContainer`, which relies on `apache/kafka` image. See gh-40695
1 parent 2ce36f9 commit 769f3e9

File tree

4 files changed

+164
-0
lines changed

4 files changed

+164
-0
lines changed
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
/*
2+
* Copyright 2012-2024 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.boot.testcontainers.service.connection.kafka;
18+
19+
import java.time.Duration;
20+
import java.util.ArrayList;
21+
import java.util.List;
22+
23+
import org.awaitility.Awaitility;
24+
import org.junit.jupiter.api.Test;
25+
import org.testcontainers.junit.jupiter.Container;
26+
import org.testcontainers.junit.jupiter.Testcontainers;
27+
import org.testcontainers.kafka.KafkaContainer;
28+
29+
import org.springframework.beans.factory.annotation.Autowired;
30+
import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
31+
import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
32+
import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
33+
import org.springframework.boot.testsupport.container.TestImage;
34+
import org.springframework.context.annotation.Bean;
35+
import org.springframework.context.annotation.Configuration;
36+
import org.springframework.kafka.annotation.KafkaListener;
37+
import org.springframework.kafka.core.KafkaTemplate;
38+
import org.springframework.test.context.TestPropertySource;
39+
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
40+
41+
import static org.assertj.core.api.Assertions.assertThat;
42+
43+
/**
44+
* Tests for {@link KafkaContainerConnectionDetailsFactory}.
45+
*
46+
* @author Moritz Halbritter
47+
* @author Andy Wilkinson
48+
* @author Phillip Webb
49+
* @author Eddú Meléndez
50+
*/
51+
@SpringJUnitConfig
52+
@Testcontainers(disabledWithoutDocker = true)
53+
@TestPropertySource(properties = { "spring.kafka.consumer.group-id=test-group",
54+
"spring.kafka.consumer.auto-offset-reset=earliest" })
55+
class ApacheKafkaContainerConnectionDetailsFactoryIntegrationTests {
56+
57+
@Container
58+
@ServiceConnection
59+
static final KafkaContainer kafka = TestImage.container(KafkaContainer.class);
60+
61+
@Autowired
62+
private KafkaTemplate<String, String> kafkaTemplate;
63+
64+
@Autowired
65+
private TestListener listener;
66+
67+
@Test
68+
void connectionCanBeMadeToKafkaContainer() {
69+
this.kafkaTemplate.send("test-topic", "test-data");
70+
Awaitility.waitAtMost(Duration.ofMinutes(4))
71+
.untilAsserted(() -> assertThat(this.listener.messages).containsExactly("test-data"));
72+
}
73+
74+
@Configuration(proxyBeanMethods = false)
75+
@ImportAutoConfiguration(KafkaAutoConfiguration.class)
76+
static class TestConfiguration {
77+
78+
@Bean
79+
TestListener testListener() {
80+
return new TestListener();
81+
}
82+
83+
}
84+
85+
static class TestListener {
86+
87+
private final List<String> messages = new ArrayList<>();
88+
89+
@KafkaListener(topics = "test-topic")
90+
void processMessage(String message) {
91+
this.messages.add(message);
92+
}
93+
94+
}
95+
96+
}
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
/*
2+
* Copyright 2012-2024 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.boot.testcontainers.service.connection.kafka;
18+
19+
import java.util.List;
20+
21+
import org.testcontainers.kafka.KafkaContainer;
22+
23+
import org.springframework.boot.autoconfigure.kafka.KafkaConnectionDetails;
24+
import org.springframework.boot.testcontainers.service.connection.ContainerConnectionDetailsFactory;
25+
import org.springframework.boot.testcontainers.service.connection.ContainerConnectionSource;
26+
import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
27+
28+
/**
29+
* {@link ContainerConnectionDetailsFactory} to create {@link KafkaConnectionDetails} from
30+
* a {@link ServiceConnection @ServiceConnection}-annotated {@link KafkaContainer}.
31+
*
32+
* @author Moritz Halbritter
33+
* @author Andy Wilkinson
34+
* @author Phillip Webb
35+
* @author Eddú Meléndez
36+
*/
37+
class ApacheKafkaContainerConnectionDetailsFactory
38+
extends ContainerConnectionDetailsFactory<KafkaContainer, KafkaConnectionDetails> {
39+
40+
@Override
41+
protected KafkaConnectionDetails getContainerConnectionDetails(ContainerConnectionSource<KafkaContainer> source) {
42+
return new KafkaContainerConnectionDetails(source);
43+
}
44+
45+
/**
46+
* {@link KafkaConnectionDetails} backed by a {@link ContainerConnectionSource}.
47+
*/
48+
private static final class KafkaContainerConnectionDetails extends ContainerConnectionDetails<KafkaContainer>
49+
implements KafkaConnectionDetails {
50+
51+
private KafkaContainerConnectionDetails(ContainerConnectionSource<KafkaContainer> source) {
52+
super(source);
53+
}
54+
55+
@Override
56+
public List<String> getBootstrapServers() {
57+
return List.of(getContainer().getBootstrapServers());
58+
}
59+
60+
}
61+
62+
}

spring-boot-project/spring-boot-testcontainers/src/main/resources/META-INF/spring.factories

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ org.springframework.boot.testcontainers.service.connection.couchbase.CouchbaseCo
1717
org.springframework.boot.testcontainers.service.connection.flyway.FlywayContainerConnectionDetailsFactory,\
1818
org.springframework.boot.testcontainers.service.connection.elasticsearch.ElasticsearchContainerConnectionDetailsFactory,\
1919
org.springframework.boot.testcontainers.service.connection.jdbc.JdbcContainerConnectionDetailsFactory,\
20+
org.springframework.boot.testcontainers.service.connection.kafka.ApacheKafkaContainerConnectionDetailsFactory,\
2021
org.springframework.boot.testcontainers.service.connection.kafka.KafkaContainerConnectionDetailsFactory,\
2122
org.springframework.boot.testcontainers.service.connection.ldap.OpenLdapContainerConnectionDetailsFactory,\
2223
org.springframework.boot.testcontainers.service.connection.liquibase.LiquibaseContainerConnectionDetailsFactory,\

spring-boot-project/spring-boot-tools/spring-boot-test-support-docker/src/main/java/org/springframework/boot/testsupport/container/TestImage.java

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,11 @@ public enum TestImage {
6464
*/
6565
ACTIVE_MQ_CLASSIC("apache/activemq-classic", "5.18.3", () -> ActiveMQContainer.class),
6666

67+
/**
68+
* A container image suitable for testing Apache Kafka.
69+
*/
70+
APACHE_KAFKA("apache/kafka", "3.7.0", () -> org.testcontainers.kafka.KafkaContainer.class),
71+
6772
/**
6873
* A container image suitable for testing Artemis.
6974
*/

0 commit comments

Comments
 (0)