From 688622798d43383175c0c245c6026ce4892dfc29 Mon Sep 17 00:00:00 2001
From: Philipp Schirmer <philipp.schirmer@bakdata.com>
Date: Thu, 30 Jan 2025 13:23:48 +0100
Subject: [PATCH] Test topic client using multiple brokers

---
 streams-bootstrap-core/build.gradle.kts               |  1 -
 .../bakdata/kafka/ApacheKafkaContainerCluster.java    | 11 +++++++----
 2 files changed, 7 insertions(+), 5 deletions(-)

diff --git a/streams-bootstrap-core/build.gradle.kts b/streams-bootstrap-core/build.gradle.kts
index a6880e13..33dc6d21 100644
--- a/streams-bootstrap-core/build.gradle.kts
+++ b/streams-bootstrap-core/build.gradle.kts
@@ -43,7 +43,6 @@ dependencies {
         name = "kafka",
         version = testContainersVersion
     )
-    testFixturesImplementation(group = "org.assertj", name = "assertj-core", version = assertJVersion)
     testImplementation(group = "io.confluent", name = "kafka-streams-avro-serde", version = confluentVersion)
     val log4jVersion: String by project
     testImplementation(group = "org.apache.logging.log4j", name = "log4j-slf4j2-impl", version = log4jVersion)
diff --git a/streams-bootstrap-core/src/testFixtures/java/com/bakdata/kafka/ApacheKafkaContainerCluster.java b/streams-bootstrap-core/src/testFixtures/java/com/bakdata/kafka/ApacheKafkaContainerCluster.java
index 87beb8e7..1ed9dcf6 100644
--- a/streams-bootstrap-core/src/testFixtures/java/com/bakdata/kafka/ApacheKafkaContainerCluster.java
+++ b/streams-bootstrap-core/src/testFixtures/java/com/bakdata/kafka/ApacheKafkaContainerCluster.java
@@ -24,7 +24,6 @@
 
 package com.bakdata.kafka;
 
-import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 
 import java.time.Duration;
@@ -98,7 +97,7 @@ public void start() {
 
         await()
                 .atMost(Duration.ofSeconds(120))
-                .untilAsserted(() -> {
+                .until(() -> {
                     final Container.ExecResult result = this.brokers.stream()
                             .findFirst()
                             .get()
@@ -111,8 +110,12 @@ public void start() {
                             );
                     final String brokers = result.getStdout().replace("\n", "");
 
-                    assertThat(brokers).asInt().isEqualTo(this.brokersNum);
-                });
+                    return Integer.parseInt(brokers);
+                }, readyBrokers -> readyBrokers == this.brokersNum);
+        await()
+                .pollDelay(Duration.ofSeconds(10))
+                .atMost(Duration.ofSeconds(11))
+                .until(() -> true);
     }
 
     @Override