Skip to content

Lazily create/set Network in KafkaContainer #2079

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Nov 22, 2019
Merged

Lazily create/set Network in KafkaContainer #2079

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2079bd3
Lazily create/set Network in KafkaContainer
bsideup Nov 17, 2019
aa1d1d4
Review fixes
bsideup Nov 20, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 32 additions & 4 deletions modules/kafka/src/main/java/org/testcontainers/containers/KafkaContainer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@
import com.github.dockerjava.api.command.InspectContainerResponse;
import com.github.dockerjava.core.command.ExecStartResultCallback;
import lombok.SneakyThrows;
import org.testcontainers.utility.Base58;
import org.testcontainers.utility.TestcontainersConfiguration;

import java.util.concurrent.TimeUnit;
Expand All @@ -25,16 +24,15 @@ public class KafkaContainer extends GenericContainer<KafkaContainer> {

private int port = PORT_NOT_ASSIGNED;

private boolean useImplicitNetwork = true;

public KafkaContainer() {
this("5.2.1");
}

public KafkaContainer(String confluentPlatformVersion) {
super(TestcontainersConfiguration.getInstance().getKafkaImage() + ":" + confluentPlatformVersion);

// TODO Only for backward compatibility
withNetwork(Network.newNetwork());
withNetworkAliases("kafka-" + Base58.randomString(6));
withExposedPorts(KAFKA_PORT);

// Use two listeners with different names, it will force Kafka to communicate with itself via internal
Expand All @@ -50,6 +48,36 @@ public KafkaContainer(String confluentPlatformVersion) {
withEnv("KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS", "0");
}

@Override
public KafkaContainer withNetwork(Network network) {
useImplicitNetwork = false;
return super.withNetwork(network);
}

@Override
public Network getNetwork() {
if (useImplicitNetwork) {
// TODO Only for backward compatibility, to be removed soon
logger().warn(
"Deprecation warning! " +
"KafkaContainer#getNetwork without an explicitly set network. " +
"Consider using KafkaContainer#withNetwork",
new Exception("Deprecated method")
);
Network network = Network.SHARED;
super.withNetwork(network);

if (getContainerId() != null) {
dockerClient.connectToNetworkCmd()
.withContainerId(getContainerId())
.withNetworkId(network.getId())
.exec();
}
return network;
}
return super.getNetwork();
}

public KafkaContainer withEmbeddedZookeeper() {
externalZookeeperConnect = null;
return self();
Expand Down
74 changes: 39 additions & 35 deletions modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,16 +31,20 @@ public void testUsage() throws Exception {
}
}

/**
* @deprecated the {@link Network} should be set explicitly with {@link KafkaContainer#withNetwork(Network)}.
*/
@Test
@Deprecated
public void testExternalZookeeperWithKafkaNetwork() throws Exception {
try (
KafkaContainer kafka = new KafkaContainer()
.withExternalZookeeper("zookeeper:2181");
KafkaContainer kafka = new KafkaContainer()
.withExternalZookeeper("zookeeper:2181");

GenericContainer zookeeper = new GenericContainer("confluentinc/cp-zookeeper:4.0.0")
.withNetwork(kafka.getNetwork())
.withNetworkAliases("zookeeper")
.withEnv("ZOOKEEPER_CLIENT_PORT", "2181");
GenericContainer zookeeper = new GenericContainer("confluentinc/cp-zookeeper:4.0.0")
.withNetwork(kafka.getNetwork())
.withNetworkAliases("zookeeper")
.withEnv("ZOOKEEPER_CLIENT_PORT", "2181");
) {
Stream.of(kafka, zookeeper).parallel().forEach(GenericContainer::start);

Expand All @@ -51,16 +55,16 @@ public void testExternalZookeeperWithKafkaNetwork() throws Exception {
@Test
public void testExternalZookeeperWithExternalNetwork() throws Exception {
try (
Network network = Network.newNetwork();
Network network = Network.newNetwork();

KafkaContainer kafka = new KafkaContainer()
.withNetwork(network)
.withExternalZookeeper("zookeeper:2181");
KafkaContainer kafka = new KafkaContainer()
.withNetwork(network)
.withExternalZookeeper("zookeeper:2181");

GenericContainer zookeeper = new GenericContainer("confluentinc/cp-zookeeper:4.0.0")
.withNetwork(network)
.withNetworkAliases("zookeeper")
.withEnv("ZOOKEEPER_CLIENT_PORT", "2181");
GenericContainer zookeeper = new GenericContainer("confluentinc/cp-zookeeper:4.0.0")
.withNetwork(network)
.withNetworkAliases("zookeeper")
.withEnv("ZOOKEEPER_CLIENT_PORT", "2181");
) {
Stream.of(kafka, zookeeper).parallel().forEach(GenericContainer::start);

Expand All @@ -70,24 +74,24 @@ public void testExternalZookeeperWithExternalNetwork() throws Exception {

protected void testKafkaFunctionality(String bootstrapServers) throws Exception {
try (
KafkaProducer<String, String> producer = new KafkaProducer<>(
ImmutableMap.of(
ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers,
ProducerConfig.CLIENT_ID_CONFIG, UUID.randomUUID().toString()
),
new StringSerializer(),
new StringSerializer()
);

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(
ImmutableMap.of(
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers,
ConsumerConfig.GROUP_ID_CONFIG, "tc-" + UUID.randomUUID(),
ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"
),
new StringDeserializer(),
new StringDeserializer()
);
KafkaProducer<String, String> producer = new KafkaProducer<>(
ImmutableMap.of(
ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers,
ProducerConfig.CLIENT_ID_CONFIG, UUID.randomUUID().toString()
),
new StringSerializer(),
new StringSerializer()
);

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(
ImmutableMap.of(
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers,
ConsumerConfig.GROUP_ID_CONFIG, "tc-" + UUID.randomUUID(),
ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"
),
new StringDeserializer(),
new StringDeserializer()
);
) {
String topicName = "messages";
consumer.subscribe(Arrays.asList(topicName));
Expand All @@ -102,9 +106,9 @@ protected void testKafkaFunctionality(String bootstrapServers) throws Exception
}

assertThat(records)
.hasSize(1)
.extracting(ConsumerRecord::topic, ConsumerRecord::key, ConsumerRecord::value)
.containsExactly(tuple(topicName, "testcontainers", "rulezzz"));
.hasSize(1)
.extracting(ConsumerRecord::topic, ConsumerRecord::key, ConsumerRecord::value)
.containsExactly(tuple(topicName, "testcontainers", "rulezzz"));

return true;
});
Expand Down