From 9ec627e4b344849caba703c4861c1696b42c264b Mon Sep 17 00:00:00 2001
From: Philipp Schirmer <philipp.schirmer@bakdata.com>
Date: Fri, 26 Jul 2024 20:55:03 +0200
Subject: [PATCH 1/2] Rename TestTopologyFactory

---
 .../src/test/java/com/bakdata/kafka/AvroMirrorTest.java       | 4 ++--
 ...BootstrapTopologyFactory.java => TestTopologyFactory.java} | 2 +-
 2 files changed, 3 insertions(+), 3 deletions(-)
 rename streams-bootstrap-test/src/main/java/com/bakdata/kafka/{StreamsBootstrapTopologyFactory.java => TestTopologyFactory.java} (99%)

diff --git a/streams-bootstrap-core/src/test/java/com/bakdata/kafka/AvroMirrorTest.java b/streams-bootstrap-core/src/test/java/com/bakdata/kafka/AvroMirrorTest.java
index ef2535be..1f3d1c6d 100644
--- a/streams-bootstrap-core/src/test/java/com/bakdata/kafka/AvroMirrorTest.java
+++ b/streams-bootstrap-core/src/test/java/com/bakdata/kafka/AvroMirrorTest.java
@@ -35,7 +35,7 @@ class AvroMirrorTest {
     private final ConfiguredStreamsApp<MirrorWithNonDefaultSerde> app = createApp();
     @RegisterExtension
     final TestTopologyExtension<TestRecord, TestRecord> testTopology =
-            StreamsBootstrapTopologyFactory.createTopologyExtensionWithSchemaRegistry(this.app);
+            TestTopologyFactory.createTopologyExtensionWithSchemaRegistry(this.app);
 
     private static ConfiguredStreamsApp<MirrorWithNonDefaultSerde> createApp() {
         final AppConfiguration<StreamsTopicConfig> configuration = new AppConfiguration<>(StreamsTopicConfig.builder()
@@ -69,7 +69,7 @@ private Serde<TestRecord> getValueSerde() {
     }
 
     private Configurator createSerdeFactory() {
-        return StreamsBootstrapTopologyFactory.createConfigurator(this.testTopology);
+        return TestTopologyFactory.createConfigurator(this.testTopology);
     }
 
     private Serde<TestRecord> getKeySerde() {
diff --git a/streams-bootstrap-test/src/main/java/com/bakdata/kafka/StreamsBootstrapTopologyFactory.java b/streams-bootstrap-test/src/main/java/com/bakdata/kafka/TestTopologyFactory.java
similarity index 99%
rename from streams-bootstrap-test/src/main/java/com/bakdata/kafka/StreamsBootstrapTopologyFactory.java
rename to streams-bootstrap-test/src/main/java/com/bakdata/kafka/TestTopologyFactory.java
index 6bc4c737..fa0a23ba 100644
--- a/streams-bootstrap-test/src/main/java/com/bakdata/kafka/StreamsBootstrapTopologyFactory.java
+++ b/streams-bootstrap-test/src/main/java/com/bakdata/kafka/TestTopologyFactory.java
@@ -35,7 +35,7 @@
  * Utility class that provides helpers for using Fluent Kafka Streams Tests with {@link ConfiguredStreamsApp}
  */
 @UtilityClass
-public class StreamsBootstrapTopologyFactory {
+public class TestTopologyFactory {
 
     /**
      * Create a {@code TestTopology} from a {@code ConfiguredStreamsApp}. It injects are {@link KafkaEndpointConfig}

From 18c9b60dd152373bdf49dc5f903c01bc8a1bd9ad Mon Sep 17 00:00:00 2001
From: Philipp Schirmer <philipp.schirmer@bakdata.com>
Date: Fri, 26 Jul 2024 20:58:13 +0200
Subject: [PATCH 2/2] Rename TestTopologyFactory

---
 .../bakdata/kafka/TestTopologyFactory.java    | 19 +++++++++----------
 1 file changed, 9 insertions(+), 10 deletions(-)

diff --git a/streams-bootstrap-test/src/main/java/com/bakdata/kafka/TestTopologyFactory.java b/streams-bootstrap-test/src/main/java/com/bakdata/kafka/TestTopologyFactory.java
index fa0a23ba..c87e8cd7 100644
--- a/streams-bootstrap-test/src/main/java/com/bakdata/kafka/TestTopologyFactory.java
+++ b/streams-bootstrap-test/src/main/java/com/bakdata/kafka/TestTopologyFactory.java
@@ -112,12 +112,7 @@ public static <K, V> TestTopologyExtension<K, V> createTopologyExtension(
      */
     public static Function<String, Map<String, Object>> getKafkaPropertiesWithSchemaRegistryUrl(
             final ConfiguredStreamsApp<? extends StreamsApp> app) {
-        return schemaRegistryUrl -> {
-            final KafkaEndpointConfig endpointConfig = newEndpointConfig()
-                    .schemaRegistryUrl(schemaRegistryUrl)
-                    .build();
-            return app.getKafkaProperties(endpointConfig);
-        };
+        return schemaRegistryUrl -> getKafkaProperties(app, schemaRegistryUrl);
     }
 
     /**
@@ -131,14 +126,18 @@ public static Configurator createConfigurator(final TestTopology<?, ?> testTopol
         return new Configurator(testTopology.getProperties());
     }
 
-    private static Map<String, Object> getKafkaProperties(final ConfiguredStreamsApp<? extends StreamsApp> app) {
-        final KafkaEndpointConfig endpointConfig = createEndpointConfig();
+    private static Map<String, Object> getKafkaProperties(final ConfiguredStreamsApp<? extends StreamsApp> app,
+            final String schemaRegistryUrl) {
+        final KafkaEndpointConfig endpointConfig = newEndpointConfig()
+                .schemaRegistryUrl(schemaRegistryUrl)
+                .build();
         return app.getKafkaProperties(endpointConfig);
     }
 
-    private static KafkaEndpointConfig createEndpointConfig() {
-        return newEndpointConfig()
+    private static Map<String, Object> getKafkaProperties(final ConfiguredStreamsApp<? extends StreamsApp> app) {
+        final KafkaEndpointConfig endpointConfig = newEndpointConfig()
                 .build();
+        return app.getKafkaProperties(endpointConfig);
     }
 
     private static KafkaEndpointConfigBuilder newEndpointConfig() {