replace zookeeper.connect to bootstrap-server due to zookeeper was de…

…precated. (#947) Bump kafka version. Co-authored-by: andrewb <[email protected]> Co-authored-by: Anastasiia Smirnova <[email protected]>
PlaytikaOSS · Sep 16, 2022 · b0eb3f9 · b0eb3f9
1 parent f89075f
commit b0eb3f9
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 5 deletions.
diff --git a/embedded-kafka/src/main/java/com/playtika/test/kafka/KafkaTopicsConfigurer.java b/embedded-kafka/src/main/java/com/playtika/test/kafka/KafkaTopicsConfigurer.java
@@ -15,6 +15,8 @@
 import java.util.Map;
 
 import static com.playtika.test.common.utils.ContainerUtils.executeInContainer;
+import static com.playtika.test.kafka.configuration.KafkaContainerConfiguration.KAFKA_HOST_NAME;
+import static java.lang.String.format;
 import static java.util.stream.Collectors.toMap;
 
 @Slf4j
@@ -59,7 +61,8 @@ public void createTopics(Collection<String> topics, Collection<TopicConfiguratio
     private void createTopic(TopicConfiguration topicConfiguration) {
         String topic = topicConfiguration.getTopic();
         int partitions = topicConfiguration.getPartitions();
-        String[] createTopicCmd = getCreateTopicCmd(topic, partitions, zookeeperProperties.getZookeeperConnect());
+        String containerBrokerList = format("%s:%d", KAFKA_HOST_NAME, kafkaProperties.getInternalBrokerPort());
+        String[] createTopicCmd = getCreateTopicCmd(topic, partitions, containerBrokerList);
         Container.ExecResult execResult = executeInContainer(this.kafka, createTopicCmd);
         log.debug("Topic={} creation cmd='{}' execResult={}", topic, createTopicCmd, execResult);
     }
@@ -80,14 +83,14 @@ private void restrictTopics(String username, Collection<String> topics) {
         }
     }
 
-    private String[] getCreateTopicCmd(String topicName, int partitions, String kafkaZookeeperConnect) {
+    private String[] getCreateTopicCmd(String topicName, int partitions, String kafkaContainerBrokerList) {
         return new String[]{
                 "kafka-topics",
                 "--create", "--topic", topicName,
                 "--partitions", String.valueOf(partitions),
                 "--replication-factor", "1",
                 "--if-not-exists",
-                "--zookeeper", kafkaZookeeperConnect
+                "--bootstrap-server", kafkaContainerBrokerList
         };
     }
 

diff --git a/...-kafka/src/main/java/com/playtika/test/kafka/properties/KafkaConfigurationProperties.java b/...-kafka/src/main/java/com/playtika/test/kafka/properties/KafkaConfigurationProperties.java
@@ -28,6 +28,7 @@ public class KafkaConfigurationProperties extends CommonContainerProperties {
 
     protected String brokerList;
     protected String containerBrokerList;
+    protected int internalBrokerPort = 9092;
     protected int brokerPort = 9093;
     protected int containerBrokerPort = 9094;
     protected int saslPlaintextBrokerPort = 9095;
@@ -69,7 +70,7 @@ public KafkaConfigurationProperties() {
     // https://docs.confluent.io/platform/current/installation/versions-interoperability.html
     @Override
     public String getDefaultDockerImage() {
-        return "confluentinc/cp-kafka:6.2.4";
+        return "confluentinc/cp-kafka:7.2.0";
     }
 
     @AllArgsConstructor

diff --git a/...c/main/java/com/playtika/test/kafka/properties/SchemaRegistryConfigurationProperties.java b/...c/main/java/com/playtika/test/kafka/properties/SchemaRegistryConfigurationProperties.java
@@ -39,7 +39,7 @@ public boolean isBasicAuthenticationEnabled() {
     // https://hub.docker.com/r/confluentinc/cp-schema-registry
     @Override
     public String getDefaultDockerImage() {
-        return "confluentinc/cp-schema-registry:6.2.0";
+        return "confluentinc/cp-schema-registry:7.2.0";
     }
 
     public enum AvroCompatibilityLevel {