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.Container;
   5 import org.testcontainers.containers.KafkaContainer;
   6 import org.testcontainers.containers.output.Slf4jLogConsumer;
   7 import org.testcontainers.containers.wait.strategy.Wait;
   8 import org.testcontainers.utility.DockerImageName;
   9
  10 import java.io.IOException;
  11
  12
  13 @Slf4j
  14 class KafkaHandoverITContainers extends AbstractHandoverITContainers
  15 {
  16   @Override
  17   void setUpExtra() throws IOException, InterruptedException
  18   {
  19     kafka.start();
  20
  21     Container.ExecResult result;
  22     result = kafka.execInContainer(
  23         "kafka-topics",
  24         "--bootstrap-server",
  25         "kafka:9999",
  26         "--create",
  27         "--topic",
  28         "info_channel",
  29         "--partitions",
  30         "3");
  31     log.info(
  32         "EXIT-CODE={}, STDOUT={}, STDERR={}",
  33         result.getExitCode(),
  34         result.getStdout(),
  35         result.getStdout());
  36     result = kafka.execInContainer(
  37         "kafka-topics",
  38         "--bootstrap-server",
  39         "kafka:9999",
  40         "--create",
  41         "--topic",
  42         "data_channel",
  43         "--partitions",
  44         "10");
  45     log.info(
  46         "EXIT-CODE={}, STDOUT={}, STDERR={}",
  47         result.getExitCode(),
  48         result.getStdout(),
  49         result.getStdout());
  50   }
  51
  52
  53   private final KafkaContainer kafka =
  54       new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.4.0"))
  55           .withNetwork(network)
  56           .withNetworkAliases("kafka")
  57           .withListener(() -> "kafka:9999")
  58           .withKraft()
  59           .waitingFor(Wait.forLogMessage(".*Kafka\\ Server\\ started.*\\n", 1))
  60           .withLogConsumer(new Slf4jLogConsumer(log, true).withPrefix("KAFKA"));
  61
  62
  63   @Override
  64   String[] getBackendCommand()
  65   {
  66     return new String[]
  67     {
  68         "--chat.backend.instance-id=backend-ID",
  69         "--chat.backend.services=kafka",
  70         "--chat.backend.kafka.bootstrap-servers=kafka:9999",
  71         "--chat.backend.kafka.instance-uri=http://backend-ID:8080",
  72         "--chat.backend.kafka.num-partitions=10",
  73         "--chat.backend.kafka.client-id-prefix=BID",
  74         "--chat.backend.kafka.haproxy-runtime-api=haproxy:8401",
  75         "--chat.backend.kafka.haproxy-map=/usr/local/etc/haproxy/sharding.map"
  76     };
  77   }
  78 }