Skip to content

Commit 8d5f7a9

Browse files
bsideupbsideup
authored
Add Kafka module (#546)
* Add Kafka module * Replace AmbassadorContainer with SocatContainer in DockerComposeContainer * make it possible to use KafkaContainer with external Zookeeper * fix typo * fix Kafka tests * Add to CHANGELOG.md, name SocatContainer, add listeners explanation comment to KafkaContainer * listen on alias * rename myNetworkAlias -> networkAlias
1 parent 35a6dde commit 8d5f7a9

12 files changed

Lines changed: 343 additions & 48 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,9 @@ All notable changes to this project will be documented in this file.
44
### Fixed
55
- Fixed retrieval of Docker host IP when running inside Docker. ([\#479](https://github.com/testcontainers/testcontainers-java/issues/479))
66

7+
### Changed
8+
- Added Kafka module ([\#546](https://github.com/testcontainers/testcontainers-java/pull/546))
9+
710
## [1.5.1] - 2017-12-19
811

912
### Fixed

‎core/src/main/java/org/testcontainers/containers/AmbassadorContainer.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,9 +15,12 @@
1515
* An ambassador container is used as a TCP proxy, enabling any TCP port of another linked container to be exposed
1616
* publicly, even if that container does not make the port public itself. The <code>richnorth/ambassador:latest</code>
1717
* container is used (based on HAProxy).
18+
*
19+
* @deprecated use {@link SocatContainer}
1820
*/
1921
@EqualsAndHashCode(callSuper = false)
2022
@Data
23+
@Deprecated
2124
public class AmbassadorContainer<SELF extends AmbassadorContainer<SELF>> extends GenericContainer<SELF> {
2225

2326
private final String otherContainerName;

‎core/src/main/java/org/testcontainers/containers/DockerComposeContainer.java‎

Lines changed: 25 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -11,9 +11,6 @@
1111
import org.apache.commons.lang.StringUtils;
1212
import org.apache.commons.lang.SystemUtils;
1313
import org.junit.runner.Description;
14-
import org.rnorth.ducttape.ratelimits.RateLimiter;
15-
import org.rnorth.ducttape.ratelimits.RateLimiterBuilder;
16-
import org.rnorth.ducttape.unreliables.Unreliables;
1714
import org.slf4j.Logger;
1815
import org.slf4j.LoggerFactory;
1916
import org.slf4j.profiler.Profiler;
@@ -27,8 +24,15 @@
2724
import org.zeroturnaround.exec.stream.slf4j.Slf4jStream;
2825

2926
import java.io.File;
30-
import java.util.*;
27+
import java.util.Arrays;
28+
import java.util.HashMap;
29+
import java.util.HashSet;
30+
import java.util.List;
31+
import java.util.Map;
32+
import java.util.Set;
33+
import java.util.concurrent.ConcurrentHashMap;
3134
import java.util.concurrent.TimeUnit;
35+
import java.util.concurrent.atomic.AtomicInteger;
3236
import java.util.stream.Collectors;
3337

3438
import static com.google.common.base.Preconditions.checkArgument;
@@ -46,7 +50,6 @@ public class DockerComposeContainer<SELF extends DockerComposeContainer<SELF>> e
4650
* Random identifier which will become part of spawned containers names, so we can shut them down
4751
*/
4852
private final String identifier;
49-
private final Map<String, AmbassadorContainer> ambassadorContainers = new HashMap<>();
5053
private final List<File> composeFiles;
5154
private final Set<String> spawnedContainerIds = new HashSet<>();
5255
private final Set<String> spawnedNetworkIds = new HashSet<>();
@@ -56,6 +59,10 @@ public class DockerComposeContainer<SELF extends DockerComposeContainer<SELF>> e
5659
private boolean pull = true;
5760
private boolean tailChildContainers;
5861

62+
private final AtomicInteger nextAmbassadorPort = new AtomicInteger(2000);
63+
private final Map<String, Map<Integer, Integer>> ambassadorPortMappings = new ConcurrentHashMap<>();
64+
private final SocatContainer ambassadorContainer = new SocatContainer();
65+
5966
private static final Object MUTEX = new Object();
6067

6168
/**
@@ -64,12 +71,6 @@ public class DockerComposeContainer<SELF extends DockerComposeContainer<SELF>> e
6471
*/
6572
private Map<String, String> env = new HashMap<>();
6673

67-
private static final RateLimiter AMBASSADOR_CREATION_RATE_LIMITER = RateLimiterBuilder
68-
.newBuilder()
69-
.withRate(6, TimeUnit.MINUTES)
70-
.withConstantThroughput()
71-
.build();
72-
7374
@Deprecated
7475
public DockerComposeContainer(File composeFile, String identifier) {
7576
this(identifier, composeFile);
@@ -201,31 +202,9 @@ private List<Container> listChildContainers() {
201202
}
202203

203204
private void startAmbassadorContainers(Profiler profiler) {
204-
for (final Map.Entry<String, AmbassadorContainer> address : ambassadorContainers.entrySet()) {
205-
206-
try {
207-
// Start any ambassador containers we need
208-
profiler.start("Ambassador container startup");
209-
210-
final AmbassadorContainer ambassadorContainer = address.getValue();
211-
Unreliables.retryUntilSuccess(120, TimeUnit.SECONDS, () -> {
212-
213-
AMBASSADOR_CREATION_RATE_LIMITER.doWhenReady(() -> {
214-
Profiler localProfiler = profiler.startNested("Ambassador container: " + ambassadorContainer.getContainerName());
215-
216-
localProfiler.start("Start ambassador container");
217-
218-
ambassadorContainer.start();
219-
});
220-
221-
return null;
222-
});
223-
} catch (Exception e) {
224-
logger().warn("Exception during ambassador container startup!", e);
225-
} finally {
226-
profiler.stop().log();
227-
}
228-
}
205+
profiler.start("Ambassador container startup");
206+
ambassadorContainer.start();
207+
profiler.stop().log();
229208
}
230209

231210
private Logger logger() {
@@ -237,8 +216,8 @@ public void finished(Description description) {
237216

238217

239218
synchronized (MUTEX) {
240-
// shut down all the ambassador containers
241-
ambassadorContainers.forEach((String address, AmbassadorContainer container) -> container.stop());
219+
// shut down the ambassador container
220+
ambassadorContainer.stop();
242221

243222
// Kill the services using docker-compose
244223
try {
@@ -270,7 +249,7 @@ public SELF withExposedService(String serviceName, int servicePort) {
270249
}
271250

272251
/*
273-
* For every service/port pair that needs to be exposed, we have to start an 'ambassador container'.
252+
* For every service/port pair that needs to be exposed, we register a target on an 'ambassador container'.
274253
*
275254
* The ambassador container's role is to link (within the Docker network) to one of the
276255
* compose services, and proxy TCP network I/O out to a port that the ambassador container
@@ -282,13 +261,12 @@ public SELF withExposedService(String serviceName, int servicePort) {
282261
* {@link GenericContainer} should ensure that the ambassador container is on the same network
283262
* as the rest of the compose environment.
284263
*/
285-
AmbassadorContainer ambassadorContainer =
286-
new AmbassadorContainer<>(new FutureContainer(this.identifier + "_" + serviceName), serviceName, servicePort)
287-
.withEnv(env);
288-
289-
// Ambassador containers will all be started together after docker compose has started
290-
ambassadorContainers.put(serviceName + ":" + servicePort, ambassadorContainer);
291264

265+
// Ambassador container will be started together after docker compose has started
266+
int ambassadorPort = nextAmbassadorPort.getAndIncrement();
267+
ambassadorPortMappings.computeIfAbsent(serviceName, __ -> new ConcurrentHashMap<>()).put(servicePort, ambassadorPort);
268+
ambassadorContainer.withTarget(ambassadorPort, serviceName, servicePort);
269+
ambassadorContainer.addLink(new FutureContainer(this.identifier + "_" + serviceName), serviceName);
292270
return self();
293271
}
294272

@@ -307,7 +285,7 @@ public DockerComposeContainer withExposedService(String serviceName, int instanc
307285
* @return a host IP address or hostname that can be used for accessing the service container.
308286
*/
309287
public String getServiceHost(String serviceName, Integer servicePort) {
310-
return ambassadorContainers.get(serviceName + ":" + servicePort).getContainerIpAddress();
288+
return ambassadorContainer.getContainerIpAddress();
311289
}
312290

313291
/**
@@ -321,7 +299,7 @@ public String getServiceHost(String serviceName, Integer servicePort) {
321299
* @return a port that can be used for accessing the service container.
322300
*/
323301
public Integer getServicePort(String serviceName, Integer servicePort) {
324-
return ambassadorContainers.get(serviceName + ":" + servicePort).getMappedPort(servicePort);
302+
return ambassadorContainer.getMappedPort(ambassadorPortMappings.get(serviceName).get(servicePort));
325303
}
326304

327305
public SELF withScaledService(String serviceBaseName, int numInstances) {

‎core/src/main/java/org/testcontainers/containers/Network.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@ class NetworkImpl extends ExternalResource implements Network {
5252
private final AtomicBoolean initialized = new AtomicBoolean();
5353

5454
@Override
55-
public String getId() {
55+
public synchronized String getId() {
5656
if (initialized.compareAndSet(false, true)) {
5757
id = create();
5858
}

‎core/src/main/java/org/testcontainers/containers/SocatContainer.java‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
package org.testcontainers.containers;
2+
3+
import org.testcontainers.utility.Base58;
4+
import org.testcontainers.utility.TestcontainersConfiguration;
5+
6+
import java.util.HashMap;
7+
import java.util.Map;
8+
import java.util.stream.Collectors;
9+
10+
/**
11+
* A socat container is used as a TCP proxy, enabling any TCP port of another container to be exposed
12+
* publicly, even if that container does not make the port public itself.
13+
*/
14+
public class SocatContainer extends GenericContainer<SocatContainer> {
15+
16+
private final Map<Integer, String> targets = new HashMap<>();
17+
18+
public SocatContainer() {
19+
super(TestcontainersConfiguration.getInstance().getSocatContainerImage());
20+
withCreateContainerCmdModifier(it -> it.withEntrypoint("/bin/sh"));
21+
withCreateContainerCmdModifier(it -> it.withName("testcontainers-socat-" + Base58.randomString(8)));
22+
}
23+
24+
public SocatContainer withTarget(int exposedPort, String host) {
25+
return withTarget(exposedPort, host, exposedPort);
26+
}
27+
28+
public SocatContainer withTarget(int exposedPort, String host, int internalPort) {
29+
addExposedPort(exposedPort);
30+
targets.put(exposedPort, String.format("%s:%s", host, internalPort));
31+
return self();
32+
}
33+
34+
@Override
35+
protected void configure() {
36+
withCommand("-c",
37+
targets.entrySet().stream()
38+
.map(entry -> "socat TCP-LISTEN:" + entry.getKey() + ",fork,reuseaddr TCP:" + entry.getValue())
39+
.collect(Collectors.joining(" & "))
40+
);
41+
}
42+
}

‎core/src/main/java/org/testcontainers/utility/TestcontainersConfiguration.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,10 @@ public String getAmbassadorContainerImage() {
3030
return (String) properties.getOrDefault("ambassador.container.image", "richnorth/ambassador:latest");
3131
}
3232

33+
public String getSocatContainerImage() {
34+
return (String) properties.getOrDefault("socat.container.image", "alpine/socat:latest");
35+
}
36+
3337
public String getVncRecordedContainerImage() {
3438
return (String) properties.getOrDefault("vncrecorder.container.image", "richnorth/vnc-recorder:latest");
3539
}

‎modules/kafka/pom.xml‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
3+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
4+
<modelVersion>4.0.0</modelVersion>
5+
6+
<parent>
7+
<groupId>org.testcontainers</groupId>
8+
<artifactId>testcontainers-parent</artifactId>
9+
<version>0-SNAPSHOT</version>
10+
<relativePath>../../pom.xml</relativePath>
11+
</parent>
12+
13+
<artifactId>kafka</artifactId>
14+
<name>TestContainers :: Apache Kafka</name>
15+
16+
<dependencies>
17+
<dependency>
18+
<groupId>${project.groupId}</groupId>
19+
<artifactId>testcontainers</artifactId>
20+
<version>${project.version}</version>
21+
</dependency>
22+
23+
<dependency>
24+
<groupId>org.apache.kafka</groupId>
25+
<artifactId>kafka-clients</artifactId>
26+
<version>1.0.0</version>
27+
<scope>test</scope>
28+
</dependency>
29+
30+
<dependency>
31+
<groupId>org.assertj</groupId>
32+
<artifactId>assertj-core</artifactId>
33+
<version>3.8.0</version>
34+
<scope>test</scope>
35+
</dependency>
36+
37+
<dependency>
38+
<groupId>com.google.guava</groupId>
39+
<artifactId>guava</artifactId>
40+
<version>23.0</version>
41+
<scope>test</scope>
42+
</dependency>
43+
44+
</dependencies>
45+
</project>

‎modules/kafka/src/main/java/org.testcontainers.containers/KafkaContainer.java‎

Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
package org.testcontainers.containers;
2+
3+
import org.testcontainers.utility.Base58;
4+
5+
import java.util.stream.Stream;
6+
7+
public class KafkaContainer extends GenericContainer<KafkaContainer> {
8+
9+
public static final int KAFKA_PORT = 9092;
10+
11+
public static final int ZOOKEEPER_PORT = 2181;
12+
13+
protected String externalZookeeperConnect = null;
14+
15+
protected SocatContainer proxy;
16+
17+
public KafkaContainer() {
18+
this("4.0.0");
19+
}
20+
21+
public KafkaContainer(String confluentPlatformVersion) {
22+
super("confluentinc/cp-kafka:" + confluentPlatformVersion);
23+
24+
withNetwork(Network.newNetwork());
25+
String networkAlias = "kafka-" + Base58.randomString(6);
26+
withNetworkAliases(networkAlias);
27+
withExposedPorts(KAFKA_PORT);
28+
29+
// Use two listeners with different names, it will force Kafka to communicate with itself via internal
30+
// listener when KAFKA_INTER_BROKER_LISTENER_NAME is set, otherwise Kafka will try to use the advertised listener
31+
withEnv("KAFKA_LISTENERS", "PLAINTEXT://0.0.0.0:9092,BROKER://" + networkAlias + ":9093");
32+
withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "BROKER:PLAINTEXT,PLAINTEXT:PLAINTEXT");
33+
withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "BROKER");
34+
35+
withEnv("KAFKA_BROKER_ID", "1");
36+
withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1");
37+
withEnv("KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS", "1");
38+
withEnv("KAFKA_LOG_FLUSH_INTERVAL_MESSAGES", Long.MAX_VALUE + "");
39+
}
40+
41+
public KafkaContainer withEmbeddedZookeeper() {
42+
externalZookeeperConnect = null;
43+
return self();
44+
}
45+
46+
public KafkaContainer withExternalZookeeper(String connectString) {
47+
externalZookeeperConnect = connectString;
48+
return self();
49+
}
50+
51+
public String getBootstrapServers() {
52+
return String.format("PLAINTEXT://%s:%s", proxy.getContainerIpAddress(), proxy.getFirstMappedPort());
53+
}
54+
55+
@Override
56+
public void start() {
57+
String networkAlias = getNetworkAliases().get(0);
58+
proxy = new SocatContainer()
59+
.withNetwork(getNetwork())
60+
.withTarget(9092, networkAlias)
61+
.withTarget(2181, networkAlias);
62+
63+
proxy.start();
64+
withEnv("KAFKA_ADVERTISED_LISTENERS", "BROKER://" + networkAlias + ":9093,PLAINTEXT://" + proxy.getContainerIpAddress() + ":" + proxy.getFirstMappedPort());
65+
66+
if (externalZookeeperConnect != null) {
67+
withEnv("KAFKA_ZOOKEEPER_CONNECT", externalZookeeperConnect);
68+
} else {
69+
addExposedPort(ZOOKEEPER_PORT);
70+
withEnv("KAFKA_ZOOKEEPER_CONNECT", "localhost:2181");
71+
withClasspathResourceMapping("tc-zookeeper.properties", "/zookeeper.properties", BindMode.READ_ONLY);
72+
withCommand("sh", "-c", "zookeeper-server-start /zookeeper.properties & /etc/confluent/docker/run");
73+
}
74+
75+
super.start();
76+
}
77+
78+
@Override
79+
public void stop() {
80+
Stream.<Runnable>of(super::stop, proxy::stop).parallel().forEach(Runnable::run);
81+
}
82+
}

‎modules/kafka/src/main/resources/tc-zookeeper.properties‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
clientPort=2181
2+
dataDir=/var/lib/zookeeper/data
3+
dataLogDir=/var/lib/zookeeper/log

0 commit comments

Comments
 (0)