Create v3

bakdata · Apr 3, 2024 · d4ef207 · d4ef207
1 parent f675e23
commit d4ef207
Show file tree

Hide file tree

Showing 29 changed files with 336 additions and 229 deletions.
diff --git a/...rap-large-messages/src/main/java/com/bakdata/kafka/LargeMessageKafkaApplicationUtils.java b/...rap-large-messages/src/main/java/com/bakdata/kafka/LargeMessageKafkaApplicationUtils.java
@@ -24,6 +24,7 @@
 
 package com.bakdata.kafka;
 
+import com.bakdata.kafka.HasTopicHooks.TopicDeletionHook;
 import java.util.Map;
 import java.util.function.Consumer;
 import lombok.experimental.UtilityClass;
@@ -41,7 +42,7 @@ public class LargeMessageKafkaApplicationUtils {
      * @return hook that cleans up LargeMessage files associated with a topic
      * @see StreamsCleanUpRunner#registerTopicDeletionHook(Consumer)
      */
-    public static Consumer<String> createLargeMessageCleanUpHook(final Map<String, Object> kafkaProperties) {
+    public static TopicDeletionHook createLargeMessageCleanUpHook(final Map<String, Object> kafkaProperties) {
         final AbstractLargeMessageConfig largeMessageConfig = new AbstractLargeMessageConfig(kafkaProperties);
         final LargeMessageStoringClient storer = largeMessageConfig.getStorer();
         return storer::deleteAllFiles;
@@ -54,22 +55,7 @@ public static Consumer<String> createLargeMessageCleanUpHook(final Map<String, O
      * @param cleanUpRunner {@code CleanUpRunner} to register hook on
      * @see #createLargeMessageCleanUpHook(Map)
      */
-    public static void registerLargeMessageCleanUpHook(final Map<String, Object> kafkaProperties,
-            final StreamsCleanUpRunner cleanUpRunner) {
-        final Consumer<String> deleteAllFiles = createLargeMessageCleanUpHook(kafkaProperties);
-        cleanUpRunner.registerTopicDeletionHook(deleteAllFiles);
-    }
-
-    /**
-     * Register a hook that cleans up LargeMessage files associated with a topic.
-     *
-     * @param kafkaProperties Kafka properties to create hook from
-     * @param cleanUpRunner {@code ProducerCleanUpRunner} to register hook on
-     * @see #createLargeMessageCleanUpHook(Map)
-     */
-    public static void registerLargeMessageCleanUpHook(final Map<String, Object> kafkaProperties,
-            final ProducerCleanUpRunner cleanUpRunner) {
-        final Consumer<String> deleteAllFiles = createLargeMessageCleanUpHook(kafkaProperties);
-        cleanUpRunner.registerTopicDeletionHook(deleteAllFiles);
+    public static void registerLargeMessageCleanUpHook(final HasTopicHooks<?> cleanUpRunner) {
+        cleanUpRunner.registerTopicDeletionHook(LargeMessageKafkaApplicationUtils::createLargeMessageCleanUpHook);
     }
 }
diff --git a/...-large-messages/src/main/java/com/bakdata/kafka/LargeMessageKafkaProducerApplication.java b/...-large-messages/src/main/java/com/bakdata/kafka/LargeMessageKafkaProducerApplication.java
@@ -30,9 +30,9 @@
 public interface LargeMessageKafkaProducerApplication extends ProducerApp {
 
     @Override
-    default void setupCleanUp(final ProducerCleanUpRunner cleanUpRunner) {
-        LargeMessageKafkaApplicationUtils.createLargeMessageCleanUpHook(cleanUpRunner.getKafkaProperties());
+    default void setupCleanUp(final ProducerCleanUpConfigurer cleanUpRunner) {
         ProducerApp.super.setupCleanUp(cleanUpRunner);
+        LargeMessageKafkaApplicationUtils.registerLargeMessageCleanUpHook(cleanUpRunner);
     }
 
 }
diff --git a/...p-large-messages/src/main/java/com/bakdata/kafka/LargeMessageKafkaStreamsApplication.java b/...p-large-messages/src/main/java/com/bakdata/kafka/LargeMessageKafkaStreamsApplication.java
@@ -30,10 +30,9 @@
 public interface LargeMessageKafkaStreamsApplication extends StreamsApp {
 
     @Override
-    default void setupCleanUp(final StreamsCleanUpRunner cleanUpRunner) {
-        LargeMessageKafkaApplicationUtils.registerLargeMessageCleanUpHook(cleanUpRunner.getKafkaProperties(),
-                cleanUpRunner);
+    default void setupCleanUp(final StreamsCleanUpConfigurer cleanUpRunner) {
         StreamsApp.super.setupCleanUp(cleanUpRunner);
+        LargeMessageKafkaApplicationUtils.registerLargeMessageCleanUpHook(cleanUpRunner);
     }
 
 }
diff --git a/streams-bootstrap-picocli/src/main/java/com/bakdata/kafka/KafkaProducerApplication.java b/streams-bootstrap-picocli/src/main/java/com/bakdata/kafka/KafkaProducerApplication.java
@@ -24,6 +24,7 @@
 
 package com.bakdata.kafka;
 
+import com.bakdata.kafka.ConfiguredProducerApp.ExecutableProducerApp;
 import java.util.Map;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
@@ -81,17 +82,21 @@ public ProducerTopicConfig createTopicConfig() {
                 .build();
     }
 
-    protected abstract ProducerApp createApp();
+    public abstract ProducerApp createApp();
 
-    private ProducerRunner createRunner() {
-        final ConfiguredProducerApp app = this.createConfiguredApp();
-        final KafkaEndpointConfig endpointConfig = this.getEndpointConfig();
-        return app.createRunner(endpointConfig);
+    public ProducerRunner createRunner() {
+        final ExecutableProducerApp executableApp = this.createExecutableApp();
+        return executableApp.createRunner();
+    }
+
+    public ProducerCleanUpRunner createCleanUpRunner() {
+        final ExecutableProducerApp executableApp = this.createExecutableApp();
+        return executableApp.createCleanUpRunner();
     }
 
-    private ProducerCleanUpRunner createCleanUpRunner() {
+    private ExecutableProducerApp createExecutableApp() {
         final ConfiguredProducerApp app = this.createConfiguredApp();
         final KafkaEndpointConfig endpointConfig = this.getEndpointConfig();
-        return app.createCleanUpRunner(endpointConfig);
+        return app.withEndpoint(endpointConfig);
     }
 }
diff --git a/streams-bootstrap-picocli/src/main/java/com/bakdata/kafka/KafkaStreamsApplication.java b/streams-bootstrap-picocli/src/main/java/com/bakdata/kafka/KafkaStreamsApplication.java
@@ -24,6 +24,7 @@
 
 package com.bakdata.kafka;
 
+import com.bakdata.kafka.ConfiguredStreamsApp.ExecutableStreamsApp;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -74,15 +75,6 @@ public abstract class KafkaStreamsApplication extends KafkaApplication implement
     @ToString.Exclude
     private List<StreamsRunner> runners = new ArrayList<>();
 
-    public static void main(final String[] args) {
-        startApplication(new KafkaStreamsApplication() {
-            @Override
-            public StreamsApp createApp() {
-                return null;
-            }
-        }, args);
-    }
-
     /**
      * Run the application. If Kafka Streams is run, this method blocks until Kafka Streams has completed shutdown,
      * either because it caught an error or the application has received a shutdown event.
@@ -124,16 +116,14 @@ public StreamsExecutionOptions createExecutionOptions() {
     }
 
     public StreamsRunner createRunner() {
-        final ConfiguredStreamsApp configuredStreamsApp = this.createConfiguredApp();
-        final KafkaEndpointConfig endpointConfig = this.getEndpointConfig();
+        final ExecutableStreamsApp executableStreamsApp = this.createExecutableApp();
         final StreamsExecutionOptions executionOptions = this.createExecutionOptions();
-        return configuredStreamsApp.createRunner(endpointConfig, executionOptions);
+        return executableStreamsApp.createRunner(executionOptions);
     }
 
     public StreamsCleanUpRunner createCleanUpRunner() {
-        final ConfiguredStreamsApp configuredStreamsApp = this.createConfiguredApp();
-        final KafkaEndpointConfig endpointConfig = this.getEndpointConfig();
-        return configuredStreamsApp.createCleanUpRunner(endpointConfig);
+        final ExecutableStreamsApp executableApp = this.createExecutableApp();
+        return executableApp.createCleanUpRunner();
     }
 
     public ConfiguredStreamsApp createConfiguredApp() {
@@ -165,6 +155,12 @@ public StreamsTopicConfig createTopicConfig() {
                 .build();
     }
 
+    public ExecutableStreamsApp createExecutableApp() {
+        final ConfiguredStreamsApp configuredStreamsApp = this.createConfiguredApp();
+        final KafkaEndpointConfig endpointConfig = this.getEndpointConfig();
+        return configuredStreamsApp.withEndpoint(endpointConfig);
+    }
+
     private StreamsOptions createStreamsOptions() {
         return StreamsOptions.builder()
                 .productive(this.productive)

diff --git a/...ams-bootstrap-picocli/src/main/java/com/bakdata/kafka/SimpleKafkaProducerApplication.java b/...ams-bootstrap-picocli/src/main/java/com/bakdata/kafka/SimpleKafkaProducerApplication.java
@@ -33,7 +33,7 @@ public class SimpleKafkaProducerApplication extends KafkaProducerApplication {
     private final @NonNull Supplier<ProducerApp> appFactory;
 
     @Override
-    protected ProducerApp createApp() {
+    public ProducerApp createApp() {
         return this.appFactory.get();
     }
 }
diff --git a/streams-bootstrap-picocli/src/test/java/com/bakdata/kafka/CliTest.java b/streams-bootstrap-picocli/src/test/java/com/bakdata/kafka/CliTest.java
@@ -56,7 +56,7 @@ void shouldExitWithSuccessCode() {
             public StreamsApp createApp() {
                 return new StreamsApp() {
                     @Override
-                    public void buildTopology(final TopologyBuilder builder, final boolean cleanUp) {
+                    public void buildTopology(final TopologyBuilder builder) {
                         throw new UnsupportedOperationException();
                     }
 
@@ -84,7 +84,7 @@ public void run() {
     void shouldExitWithErrorCodeOnRunError() {
         KafkaApplication.startApplication(new SimpleKafkaStreamsApplication(() -> new StreamsApp() {
             @Override
-            public void buildTopology(final TopologyBuilder builder, final boolean cleanUp) {
+            public void buildTopology(final TopologyBuilder builder) {
                 throw new UnsupportedOperationException();
             }
 
@@ -108,7 +108,7 @@ void shouldExitWithErrorCodeOnCleanupError() {
             public StreamsApp createApp() {
                 return new StreamsApp() {
                     @Override
-                    public void buildTopology(final TopologyBuilder builder, final boolean cleanUp) {
+                    public void buildTopology(final TopologyBuilder builder) {
                         throw new UnsupportedOperationException();
                     }
 
@@ -140,7 +140,7 @@ void shouldExitWithErrorCodeOnMissingBrokerParameter() {
             public StreamsApp createApp() {
                 return new StreamsApp() {
                     @Override
-                    public void buildTopology(final TopologyBuilder builder, final boolean cleanUp) {
+                    public void buildTopology(final TopologyBuilder builder) {
                         throw new UnsupportedOperationException();
                     }
 
@@ -169,7 +169,7 @@ void shouldExitWithErrorInTopology() throws InterruptedException {
         try (final EmbeddedKafkaCluster kafkaCluster = provisionWith(defaultClusterConfig());
                 final KafkaStreamsApplication app = new SimpleKafkaStreamsApplication(() -> new StreamsApp() {
                     @Override
-                    public void buildTopology(final TopologyBuilder builder, final boolean cleanUp) {
+                    public void buildTopology(final TopologyBuilder builder) {
                         builder.streamInput(Consumed.with(Serdes.ByteArray(), Serdes.ByteArray()))
                                 .peek((k, v) -> {
                                     throw new RuntimeException();
@@ -202,7 +202,7 @@ void shouldExitWithSuccessCodeOnShutdown() throws InterruptedException {
         try (final EmbeddedKafkaCluster kafkaCluster = provisionWith(defaultClusterConfig());
                 final KafkaStreamsApplication app = new SimpleKafkaStreamsApplication(() -> new StreamsApp() {
                     @Override
-                    public void buildTopology(final TopologyBuilder builder, final boolean cleanUp) {
+                    public void buildTopology(final TopologyBuilder builder) {
                         builder.streamInput(Consumed.with(Serdes.ByteArray(), Serdes.ByteArray()))
                                 .to(builder.getTopics().getOutputTopic());
                     }
@@ -242,7 +242,7 @@ void shouldExitWithErrorOnCleanupError() {
             public StreamsApp createApp() {
                 return new StreamsApp() {
                     @Override
-                    public void buildTopology(final TopologyBuilder builder, final boolean cleanUp) {
+                    public void buildTopology(final TopologyBuilder builder) {
                         throw new UnsupportedOperationException();
                     }
 
@@ -268,7 +268,7 @@ void shouldParseArguments() {
             public StreamsApp createApp() {
                 return new StreamsApp() {
                     @Override
-                    public void buildTopology(final TopologyBuilder builder, final boolean cleanUp) {
+                    public void buildTopology(final TopologyBuilder builder) {
                         throw new UnsupportedOperationException();
                     }
 

diff --git a/streams-bootstrap-picocli/src/test/java/com/bakdata/kafka/test_applications/Mirror.java b/streams-bootstrap-picocli/src/test/java/com/bakdata/kafka/test_applications/Mirror.java
@@ -33,7 +33,7 @@
 @NoArgsConstructor
 public class Mirror implements StreamsApp {
     @Override
-    public void buildTopology(final TopologyBuilder builder, final boolean cleanUp) {
+    public void buildTopology(final TopologyBuilder builder) {
         final KStream<String, String> input = builder.streamInput();
         input.to(builder.getTopics().getOutputTopic());
     }

diff --git a/streams-bootstrap-picocli/src/test/java/com/bakdata/kafka/test_applications/WordCount.java b/streams-bootstrap-picocli/src/test/java/com/bakdata/kafka/test_applications/WordCount.java
@@ -40,7 +40,7 @@
 public class WordCount implements StreamsApp {
 
     @Override
-    public void buildTopology(final TopologyBuilder builder, final boolean cleanUp) {
+    public void buildTopology(final TopologyBuilder builder) {
         final KStream<String, String> textLines = builder.streamInput();
 
         final Pattern pattern = Pattern.compile("\\W+", Pattern.UNICODE_CHARACTER_CLASS);

diff --git a/streams-bootstrap/src/main/java/com/bakdata/kafka/ConfiguredProducerApp.java b/streams-bootstrap/src/main/java/com/bakdata/kafka/ConfiguredProducerApp.java
@@ -87,21 +87,31 @@ public Map<String, Object> getKafkaProperties(final KafkaEndpointConfig endpoint
         return kafkaConfig;
     }
 
-    public ProducerCleanUpRunner createCleanUpRunner(final KafkaEndpointConfig endpointConfig) {
-        final Map<String, Object> kafkaProperties = this.getKafkaProperties(endpointConfig);
-        final ProducerCleanUpRunner cleanUpRunner =
-                new ProducerCleanUpRunner(this.configuration.getTopics(), kafkaProperties);
+    public ExecutableProducerApp withEndpoint(final KafkaEndpointConfig endpointConfig) {
+        return new ExecutableProducerApp(endpointConfig);
+    }
+
+    @RequiredArgsConstructor
+    public class ExecutableProducerApp {
+        private final @NonNull KafkaEndpointConfig endpointConfig;
 
-        this.app.setupCleanUp(cleanUpRunner);
-        return cleanUpRunner;
+        public ProducerCleanUpRunner createCleanUpRunner() {
+            final Map<String, Object> kafkaProperties =
+                    ConfiguredProducerApp.this.getKafkaProperties(this.endpointConfig);
+            final ProducerCleanUpConfigurer configurer = new ProducerCleanUpConfigurer();
+            ConfiguredProducerApp.this.app.setupCleanUp(configurer);
+            return ProducerCleanUpRunner.create(ConfiguredProducerApp.this.configuration.getTopics(), kafkaProperties,
+                    configurer);
     }
 
-    public ProducerRunner createRunner(final KafkaEndpointConfig endpointConfig) {
-        final Map<String, Object> kafkaProperties = this.getKafkaProperties(endpointConfig);
+        public ProducerRunner createRunner() {
+            final Map<String, Object> kafkaProperties =
+                    ConfiguredProducerApp.this.getKafkaProperties(this.endpointConfig);
         final ProducerBuilder producerBuilder = ProducerBuilder.builder()
-                .topics(this.configuration.getTopics())
+                .topics(ConfiguredProducerApp.this.configuration.getTopics())
                 .kafkaProperties(kafkaProperties)
                 .build();
-        return new ProducerRunner(() -> this.app.run(producerBuilder));
+            return new ProducerRunner(() -> ConfiguredProducerApp.this.app.run(producerBuilder));
+        }
     }
 }
diff --git a/streams-bootstrap/src/main/java/com/bakdata/kafka/ConfiguredStreamsApp.java b/streams-bootstrap/src/main/java/com/bakdata/kafka/ConfiguredStreamsApp.java
@@ -97,30 +97,8 @@ public Topology createTopology(final Map<String, Object> kafkaProperties) {
         return this.createTopology(kafkaProperties, false);
     }
 
-    public StreamsCleanUpRunner createCleanUpRunner(final KafkaEndpointConfig endpointConfig) {
-        final Map<String, Object> kafkaProperties = this.getKafkaProperties(endpointConfig);
-        final Topology topology = this.createTopology(kafkaProperties, true);
-        final StreamsCleanUpRunner cleanUpRunner = StreamsCleanUpRunner.create(topology, kafkaProperties);
-        cleanUpRunner.registerFinishHook(this.app::close);
-
-        this.app.setupCleanUp(cleanUpRunner);
-        return cleanUpRunner;
-    }
-
-    public StreamsRunner createRunner(final KafkaEndpointConfig endpointConfig) {
-        return this.createRunner(endpointConfig, StreamsExecutionOptions.builder().build());
-    }
-
-    public StreamsRunner createRunner(final KafkaEndpointConfig endpointConfig,
-            final StreamsExecutionOptions executionOptions) {
-        final Map<String, Object> kafkaProperties = this.getKafkaProperties(endpointConfig);
-        final Topology topology = this.createTopology(kafkaProperties);
-        return StreamsRunner.builder()
-                .topology(topology)
-                .config(new StreamsConfig(kafkaProperties))
-                .executionOptions(executionOptions)
-                .hooks(this.createHooks())
-                .build();
+    public ExecutableStreamsApp withEndpoint(final KafkaEndpointConfig endpointConfig) {
+        return new ExecutableStreamsApp(endpointConfig);
     }
 
     /**
@@ -134,18 +112,52 @@ private Topology createTopology(final Map<String, Object> kafkaProperties, final
         final TopologyBuilder topologyBuilder = TopologyBuilder.builder()
                 .topics(this.configuration.getTopics())
                 .kafkaProperties(kafkaProperties)
+                .cleanUp(cleanUp)
                 .build();
-        this.app.buildTopology(topologyBuilder, cleanUp);
+        this.app.buildTopology(topologyBuilder);
         return topologyBuilder.build();
     }
 
-    private StreamsHooks createHooks() {
-        return StreamsHooks.builder()
-                .stateListener(this.app.getStateListener())
-                .uncaughtExceptionHandler(this.app.getUncaughtExceptionHandler())
-                .onStart(this.app::onStreamsStart)
-                .onShutdown(this.app::close)
-                .build();
+    @RequiredArgsConstructor
+    public class ExecutableStreamsApp {
+
+        private final @NonNull KafkaEndpointConfig endpointConfig;
+
+        public StreamsCleanUpRunner createCleanUpRunner() {
+            final Map<String, Object> kafkaProperties =
+                    ConfiguredStreamsApp.this.getKafkaProperties(this.endpointConfig);
+            final Topology topology = ConfiguredStreamsApp.this.createTopology(kafkaProperties, true);
+            final StreamsCleanUpConfigurer configurer = new StreamsCleanUpConfigurer();
+            ConfiguredStreamsApp.this.app.setupCleanUp(configurer);
+            configurer.registerFinishHook(ConfiguredStreamsApp.this.app::close);
+            return StreamsCleanUpRunner.create(topology, kafkaProperties, configurer);
+        }
+
+        public StreamsRunner createRunner() {
+            return this.createRunner(StreamsExecutionOptions.builder().build());
+        }
+
+        public StreamsRunner createRunner(final StreamsExecutionOptions executionOptions) {
+            final Map<String, Object> kafkaProperties =
+                    ConfiguredStreamsApp.this.getKafkaProperties(this.endpointConfig);
+            final Topology topology = ConfiguredStreamsApp.this.createTopology(kafkaProperties);
+            return StreamsRunner.builder()
+                    .topology(topology)
+                    .config(new StreamsConfig(kafkaProperties))
+                    .executionOptions(executionOptions)
+                    .hooks(this.createHooks())
+                    .build();
+        }
+
+        private StreamsHooks createHooks() {
+            return StreamsHooks.builder()
+                    .stateListener(ConfiguredStreamsApp.this.app.getStateListener())
+                    .uncaughtExceptionHandler(ConfiguredStreamsApp.this.app.getUncaughtExceptionHandler())
+                    .onStart(ConfiguredStreamsApp.this.app::onStreamsStart)
+                    .onShutdown(ConfiguredStreamsApp.this.app::close)
+                    .build();
+        }
+
     }
 
 }