testcontainers-java/modules/kafka/src/test/java/org/testcontainers/kafka/ConfluentKafkaContainerTest.java at 96376db1b671ad5e38951a51d1c9624a8313ee0a · julianladisch/testcontainers-java · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
package org.testcontainers.kafka;

import com.github.dockerjava.api.command.InspectContainerResponse;
import lombok.SneakyThrows;
import org.junit.Test;
import org.testcontainers.AbstractKafka;
import org.testcontainers.KCatContainer;
import org.testcontainers.containers.Network;
import org.testcontainers.containers.SocatContainer;
import org.testcontainers.utility.MountableFile;

public class ConfluentKafkaContainerTest extends AbstractKafka {

    @Test
    public void testUsage() throws Exception {
        try ( // constructorWithVersion {
            ConfluentKafkaContainer kafka = new ConfluentKafkaContainer("confluentinc/cp-kafka:7.4.0")
            // }
        ) {
            kafka.start();
            testKafkaFunctionality(kafka.getBootstrapServers());
        }
    }

    @Test
    public void testUsageWithNetworkAlias() {
        try (
            Network network = Network.newNetwork();
            ConfluentKafkaContainer kafka = new ConfluentKafkaContainer("confluentinc/cp-kafka:7.4.0")
                .withNetworkAliases("mykafka")
                .withNetwork(network)
        ) {
            kafka.start();
            assertKafka("mykafka:9092", network);
        }
    }

    @Test
    public void testUsageWithListener() throws Exception {
        try (
            Network network = Network.newNetwork();
            // registerListener {
            ConfluentKafkaContainer kafka = new ConfluentKafkaContainer("confluentinc/cp-kafka:7.4.0")
                .withListener("kafka:19092")
                .withNetwork(network);
            // }
            KCatContainer kcat = new KCatContainer().withNetwork(network)
        ) {
            kafka.start();
            assertKafka("kafka:19092", network);
        }
    }

    @Test
    public void testUsageWithListenerFromProxy() throws Exception {
        try (
            Network network = Network.newNetwork();
            // registerListenerFromProxy {
            SocatContainer socat = new SocatContainer().withNetwork(network).withTarget(2000, "kafka", 19092);
            ConfluentKafkaContainer kafka = new ConfluentKafkaContainer("confluentinc/cp-kafka:7.4.0")
                .withListener("kafka:19092", () -> socat.getHost() + ":" + socat.getMappedPort(2000))
                .withNetwork(network)
            // }
        ) {
            socat.start();
            kafka.start();

            String bootstrapServers = String.format("%s:%s", socat.getHost(), socat.getMappedPort(2000));
            testKafkaFunctionality(bootstrapServers);
        }
    }

    @SneakyThrows
    @Test
    public void shouldConfigureAuthenticationWithSaslUsingJaas() {
        try (
            ConfluentKafkaContainer kafka = new ConfluentKafkaContainer("confluentinc/cp-kafka:7.7.0")
                .withEnv(
                    "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP",
                    "PLAINTEXT:SASL_PLAINTEXT,BROKER:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT"
                )
                .withEnv("KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL", "PLAIN")
                .withEnv("KAFKA_LISTENER_NAME_PLAINTEXT_SASL_ENABLED_MECHANISMS", "PLAIN")
                .withEnv("KAFKA_LISTENER_NAME_BROKER_SASL_ENABLED_MECHANISMS", "PLAIN")
                .withEnv("KAFKA_LISTENER_NAME_BROKER_PLAIN_SASL_JAAS_CONFIG", getJaasConfig())
                .withEnv("KAFKA_LISTENER_NAME_PLAINTEXT_PLAIN_SASL_JAAS_CONFIG", getJaasConfig())
        ) {
            kafka.start();

            testSecurePlainKafkaFunctionality(kafka.getBootstrapServers());
        }
    }

    @SneakyThrows
    @Test
    public void shouldConfigureAuthenticationWithSaslScramUsingJaas() {
        try (
            ConfluentKafkaContainer kafka = new ConfluentKafkaContainer("confluentinc/cp-kafka:7.7.0") {
                @SneakyThrows
                @Override
                protected void containerIsStarting(InspectContainerResponse containerInfo) {
                    String command =
                        "echo 'kafka-storage format --ignore-formatted -t \"" +
                        "$CLUSTER_ID" +
                        "\" --add-scram SCRAM-SHA-256=[name=admin,password=admin] -c /etc/kafka/kafka.properties' >> /etc/confluent/docker/configure";
                    execInContainer("bash", "-c", command);
                    super.containerIsStarting(containerInfo);
                }
            }
                .withEnv(
                    "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP",
                    "PLAINTEXT:SASL_PLAINTEXT,BROKER:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT"
                )
                .withEnv("KAFKA_LISTENER_NAME_PLAINTEXT_SASL_ENABLED_MECHANISMS", "SCRAM-SHA-256")
                .withEnv("KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL", "SCRAM-SHA-256")
                .withEnv("KAFKA_SASL_ENABLED_MECHANISMS", "SCRAM-SHA-256")
                .withEnv("KAFKA_OPTS", "-Djava.security.auth.login.config=/etc/kafka/secrets/kafka_server_jaas.conf")
                .withCopyFileToContainer(
                    MountableFile.forClasspathResource("kafka_server_jaas.conf"),
                    "/etc/kafka/secrets/kafka_server_jaas.conf"
                )
        ) {
            kafka.start();

            testSecureScramKafkaFunctionality(kafka.getBootstrapServers());
        }
    }
}