1 package de.juplo.kafka.chat.backend;
3 import lombok.extern.slf4j.Slf4j;
4 import org.awaitility.Awaitility;
5 import org.junit.jupiter.api.BeforeAll;
6 import org.junit.jupiter.api.BeforeEach;
7 import org.springframework.http.HttpStatus;
8 import org.springframework.web.reactive.function.client.WebClient;
9 import org.testcontainers.containers.*;
10 import org.testcontainers.containers.output.Slf4jLogConsumer;
11 import org.testcontainers.containers.wait.strategy.Wait;
12 import org.testcontainers.utility.DockerImageName;
13 import reactor.core.publisher.Mono;
15 import java.io.IOException;
16 import java.time.Duration;
20 class KafkaHandoverIT extends AbstractHandoverIT
25 Integer port = HAPROXY.getMappedPort(8400);
26 webClient = WebClient.create("http://localhost:" + port);
30 .atMost(Duration.ofMinutes(10))
31 .until(() -> WebClient
32 .create("http://localhost:" + BACKEND_1.getMappedPort(8080))
34 .uri("/actuator/health")
35 .exchangeToMono(response ->
37 if (response.statusCode().equals(HttpStatus.OK))
40 .bodyToMono(StatusTo.class)
41 .map(StatusTo::getStatus)
42 .map(status -> status.equalsIgnoreCase("UP"));
46 return Mono.just(false);
53 .killContainerCmd(HAPROXY.getContainerId())
60 .atMost(Duration.ofMinutes(10))
61 .until(() -> webClient
63 .uri("/actuator/health")
64 .exchangeToMono(response ->
66 if (response.statusCode().equals(HttpStatus.OK))
69 .bodyToMono(StatusTo.class)
70 .map(StatusTo::getStatus)
71 .map(status -> status.equalsIgnoreCase("UP"));
75 return Mono.just(false);
83 static void setUpDocker() throws IOException, InterruptedException
88 Container.ExecResult result;
89 result = KAFKA.execInContainer(
99 "EXIT-CODE={}, STDOUT={}, STDERR={}",
100 result.getExitCode(),
103 result = KAFKA.execInContainer(
105 "--bootstrap-server",
113 "EXIT-CODE={}, STDOUT={}, STDERR={}",
114 result.getExitCode(),
119 // BACKEND_2.start();
120 // BACKEND_3.start();
123 static Network NETWORK = Network.newNetwork();
125 static KafkaContainer KAFKA =
126 new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.4.0"))
127 .withNetwork(NETWORK)
128 .withNetworkAliases("kafka")
129 .withListener(() -> "kafka:9999")
131 .waitingFor(Wait.forLogMessage(".*Kafka\\ Server\\ started.*\\n", 1))
132 .withLogConsumer(new Slf4jLogConsumer(log, true).withPrefix("KAFKA"));
134 static GenericContainer BACKEND_1 =
135 new GenericContainer(DockerImageName.parse("juplo/chat-backend:0.0.1-SNAPSHOT"))
136 .withImagePullPolicy(NEVER_PULL)
137 .withNetwork(NETWORK)
138 .withNetworkAliases("backend-1")
140 "--chat.backend.instance-id=backend-1",
141 "--chat.backend.services=kafka",
142 "--chat.backend.kafka.bootstrap-servers=kafka:9999",
143 "--chat.backend.kafka.instance-uri=http://backend-1:8080",
144 "--chat.backend.kafka.num-partitions=10",
145 "--chat.backend.kafka.client-id-prefix=B1",
146 "--chat.backend.kafka.haproxy-runtime-api=haproxy:8401",
147 "--chat.backend.kafka.haproxy-map=/usr/local/etc/haproxy/sharding.map"
149 .withExposedPorts(8080)
151 .waitingFor(Wait.forLogMessage(".*Started\\ ChatBackendApplication.*\\n", 1))
152 .withLogConsumer(new Slf4jLogConsumer(log, true).withPrefix("BACKEND-1"));
154 static GenericContainer BACKEND_2 =
155 new GenericContainer(DockerImageName.parse("juplo/chat-backend:0.0.1-SNAPSHOT"))
156 .withImagePullPolicy(NEVER_PULL)
157 .withNetwork(NETWORK)
158 .withNetworkAliases("backend-2")
160 "--chat.backend.instance-id=backend-2",
161 "--chat.backend.services=kafka",
162 "--chat.backend.kafka.bootstrap-servers=kafka:9999",
163 "--chat.backend.kafka.instance-uri=http://backend-2:8080",
164 "--chat.backend.kafka.num-partitions=10",
165 "--chat.backend.kafka.client-id-prefix=B2",
166 "--chat.backend.kafka.haproxy-runtime-api=haproxy:8401",
167 "--chat.backend.kafka.haproxy-map=/usr/local/etc/haproxy/sharding.map"
169 .withExposedPorts(8080)
171 .waitingFor(Wait.forLogMessage(".*Started\\ ChatBackendApplication.*\\n", 1))
172 .withLogConsumer(new Slf4jLogConsumer(log, true).withPrefix("BACKEND-2"));
175 static GenericContainer BACKEND_3 =
176 new GenericContainer(DockerImageName.parse("juplo/chat-backend:0.0.1-SNAPSHOT"))
177 .withImagePullPolicy(NEVER_PULL)
178 .withNetwork(NETWORK)
179 .withNetworkAliases("backend-3")
181 "--chat.backend.instance-id=backend-3",
182 "--chat.backend.services=kafka",
183 "--chat.backend.kafka.bootstrap-servers=kafka:9999",
184 "--chat.backend.kafka.instance-uri=http://backend-3:8080",
185 "--chat.backend.kafka.num-partitions=10",
186 "--chat.backend.kafka.client-id-prefix=B3",
187 "--chat.backend.kafka.haproxy-runtime-api=haproxy:8401",
188 "--chat.backend.kafka.haproxy-map=/usr/local/etc/haproxy/sharding.map"
190 .withExposedPorts(8080)
192 .waitingFor(Wait.forLogMessage(".*Started\\ ChatBackendApplication.*\\n", 1))
193 .withLogConsumer(new Slf4jLogConsumer(log, true).withPrefix("BACKEND-3"));
195 static GenericContainer HAPROXY =
196 new GenericContainer(DockerImageName.parse("haproxytech/haproxy-debian:2.8"))
197 .withNetwork(NETWORK)
198 .withNetworkAliases("haproxy")
199 .withClasspathResourceMapping(
201 "/usr/local/etc/haproxy/haproxy.cfg",
203 .withClasspathResourceMapping(
205 "/usr/local/etc/haproxy/sharding.map",
207 .withExposedPorts(8400, 8401, 8404)
208 .withLogConsumer(new Slf4jLogConsumer(log, true).withPrefix("HAPROXY"));