Create v3

bakdata · Apr 19, 2024 · 96e2dad · 96e2dad
1 parent 38aa137
commit 96e2dad
Show file tree

Hide file tree

Showing 3 changed files with 25 additions and 12 deletions.
diff --git a/...ams-bootstrap-large-messages/src/main/java/com/bakdata/kafka/LargeMessageProducerApp.java b/...ams-bootstrap-large-messages/src/main/java/com/bakdata/kafka/LargeMessageProducerApp.java
@@ -29,17 +29,24 @@
  */
 public interface LargeMessageProducerApp extends ProducerApp {
 
-    static ProducerCleanUpConfiguration registerLargeMessageCleanUpHook(final ProducerCleanUpConfiguration configurer,
-            final EffectiveAppConfiguration<?> configuration) {
-        return configurer.registerTopicHook(
+    /**
+     * Register a hook that cleans up LargeMessage files associated with a topic
+     * @param cleanUpConfiguration Configuration to register hook on
+     * @param configuration Configuration to create hook from
+     * @return {@code ProducerCleanUpConfiguration} with registered topic hook
+     * @see LargeMessageKafkaApplicationUtils#createLargeMessageCleanUpHook(EffectiveAppConfiguration)
+     */
+    static ProducerCleanUpConfiguration registerLargeMessageCleanUpHook(
+            final ProducerCleanUpConfiguration cleanUpConfiguration, final EffectiveAppConfiguration<?> configuration) {
+        return cleanUpConfiguration.registerTopicHook(
                 LargeMessageKafkaApplicationUtils.createLargeMessageCleanUpHook(configuration));
     }
 
     @Override
     default ProducerCleanUpConfiguration setupCleanUp(
             final EffectiveAppConfiguration<ProducerTopicConfig> configuration) {
-        final ProducerCleanUpConfiguration configurer = ProducerApp.super.setupCleanUp(configuration);
-        return registerLargeMessageCleanUpHook(configurer, configuration);
+        final ProducerCleanUpConfiguration cleanUpConfiguration = ProducerApp.super.setupCleanUp(configuration);
+        return registerLargeMessageCleanUpHook(cleanUpConfiguration, configuration);
     }
 
 }
diff --git a/streams-bootstrap-large-messages/src/main/java/com/bakdata/kafka/LargeMessageStreamsApp.java b/streams-bootstrap-large-messages/src/main/java/com/bakdata/kafka/LargeMessageStreamsApp.java
@@ -29,17 +29,24 @@
  */
 public interface LargeMessageStreamsApp extends StreamsApp {
 
-    static StreamsCleanUpConfiguration registerLargeMessageCleanUpHook(final StreamsCleanUpConfiguration configurer,
-            final EffectiveAppConfiguration<?> configuration) {
-        return configurer.registerTopicHook(
+    /**
+     * Register a hook that cleans up LargeMessage files associated with a topic
+     * @param cleanUpConfiguration Configuration to register hook on
+     * @param configuration Configuration to create hook from
+     * @return {@code StreamsCleanUpConfiguration} with registered topic hook
+     * @see LargeMessageKafkaApplicationUtils#createLargeMessageCleanUpHook(EffectiveAppConfiguration)
+     */
+    static StreamsCleanUpConfiguration registerLargeMessageCleanUpHook(
+            final StreamsCleanUpConfiguration cleanUpConfiguration, final EffectiveAppConfiguration<?> configuration) {
+        return cleanUpConfiguration.registerTopicHook(
                 LargeMessageKafkaApplicationUtils.createLargeMessageCleanUpHook(configuration));
     }
 
     @Override
     default StreamsCleanUpConfiguration setupCleanUp(
             final EffectiveAppConfiguration<StreamsTopicConfig> configuration) {
-        final StreamsCleanUpConfiguration configurer = StreamsApp.super.setupCleanUp(configuration);
-        return registerLargeMessageCleanUpHook(configurer, configuration);
+        final StreamsCleanUpConfiguration cleanUpConfiguration = StreamsApp.super.setupCleanUp(configuration);
+        return registerLargeMessageCleanUpHook(cleanUpConfiguration, configuration);
     }
 
 }
diff --git a/streams-bootstrap-picocli/src/main/java/com/bakdata/kafka/KafkaApplication.java b/streams-bootstrap-picocli/src/main/java/com/bakdata/kafka/KafkaApplication.java
@@ -232,8 +232,7 @@ public final CA createConfiguredApp(final boolean cleanUp) {
      */
     public final AppConfiguration<T> createConfiguration() {
         final T topics = this.createTopicConfig();
-        final Map<String, String> kafkaConfig = this.getKafkaConfig();
-        return new AppConfiguration<>(topics, kafkaConfig);
+        return new AppConfiguration<>(topics, this.kafkaConfig);
     }
 
     /**