juplo.de Git - demos/kafka/chat/blob - src/test/java/de/juplo/kafka/chat/backend/KafkaHandoverITContainers.java

   1 package de.juplo.kafka.chat.backend;
   2
   3 import lombok.extern.slf4j.Slf4j;
   4 import org.testcontainers.containers.*;
   5 import org.testcontainers.containers.output.Slf4jLogConsumer;
   6 import org.testcontainers.containers.wait.strategy.Wait;
   7 import org.testcontainers.utility.DockerImageName;
   8
   9 import java.io.IOException;
  10
  11
  12 @Slf4j
  13 class KafkaHandoverIT extends AbstractHandoverIT
  14 {
  15   @Override
  16   void setUpExtra() throws IOException, InterruptedException
  17   {
  18     kafka.start();
  19
  20     Container.ExecResult result;
  21     result = kafka.execInContainer(
  22         "kafka-topics",
  23         "--bootstrap-server",
  24         "kafka:9999",
  25         "--create",
  26         "--topic",
  27         "info_channel",
  28         "--partitions",
  29         "3");
  30     log.info(
  31         "EXIT-CODE={}, STDOUT={}, STDERR={}",
  32         result.getExitCode(),
  33         result.getStdout(),
  34         result.getStdout());
  35     result = kafka.execInContainer(
  36         "kafka-topics",
  37         "--bootstrap-server",
  38         "kafka:9999",
  39         "--create",
  40         "--topic",
  41         "data_channel",
  42         "--partitions",
  43         "10");
  44     log.info(
  45         "EXIT-CODE={}, STDOUT={}, STDERR={}",
  46         result.getExitCode(),
  47         result.getStdout(),
  48         result.getStdout());
  49   }
  50
  51
  52   KafkaContainer kafka =
  53       new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.4.0"))
  54           .withNetwork(network)
  55           .withNetworkAliases("kafka")
  56           .withListener(() -> "kafka:9999")
  57           .withKraft()
  58           .waitingFor(Wait.forLogMessage(".*Kafka\\ Server\\ started.*\\n", 1))
  59           .withLogConsumer(new Slf4jLogConsumer(log, true).withPrefix("KAFKA"));
  60
  61   @Override
  62   String[] getCommandBackend1()
  63   {
  64     return new String[]
  65     {
  66         "--chat.backend.instance-id=backend-1",
  67         "--chat.backend.services=kafka",
  68         "--chat.backend.kafka.bootstrap-servers=kafka:9999",
  69         "--chat.backend.kafka.instance-uri=http://backend-1:8080",
  70         "--chat.backend.kafka.num-partitions=10",
  71         "--chat.backend.kafka.client-id-prefix=B1",
  72         "--chat.backend.kafka.haproxy-runtime-api=haproxy:8401",
  73         "--chat.backend.kafka.haproxy-map=/usr/local/etc/haproxy/sharding.map"
  74     };
  75   }
  76
  77   @Override
  78   String[] getCommandBackend2()
  79   {
  80     return new String[]
  81     {
  82         "--chat.backend.instance-id=backend-2",
  83         "--chat.backend.services=kafka",
  84         "--chat.backend.kafka.bootstrap-servers=kafka:9999",
  85         "--chat.backend.kafka.instance-uri=http://backend-2:8080",
  86         "--chat.backend.kafka.num-partitions=10",
  87         "--chat.backend.kafka.client-id-prefix=B2",
  88         "--chat.backend.kafka.haproxy-runtime-api=haproxy:8401",
  89         "--chat.backend.kafka.haproxy-map=/usr/local/etc/haproxy/sharding.map"
  90     };
  91   }
  92
  93   @Override
  94   String[] getCommandBackend3()
  95   {
  96     return new String[]
  97     {
  98         "--chat.backend.instance-id=backend-3",
  99         "--chat.backend.services=kafka",
 100         "--chat.backend.kafka.bootstrap-servers=kafka:9999",
 101         "--chat.backend.kafka.instance-uri=http://backend-3:8080",
 102         "--chat.backend.kafka.num-partitions=10",
 103         "--chat.backend.kafka.client-id-prefix=B3",
 104         "--chat.backend.kafka.haproxy-runtime-api=haproxy:8401",
 105         "--chat.backend.kafka.haproxy-map=/usr/local/etc/haproxy/sharding.map"
 106     };
 107   }
 108 }