Create v3

streams-bootstrap/src/main/java/com/bakdata/kafka/ConfiguredStreamsApp.java

@@ -175,7 +175,7 @@ public void close() {
     }
 
     private void setupApp(final Map<String, Object> kafkaProperties) {
-        final StreamsAppSetupConfiguration setupConfiguration = StreamsAppSetupConfiguration.builder()
+        final StreamsSetupConfiguration setupConfiguration = StreamsSetupConfiguration.builder()
                 .kafkaProperties(kafkaProperties)
                 .topics(this.getTopics())
                 .build();

streams-bootstrap/src/main/java/com/bakdata/kafka/ExecutableProducerApp.java

@@ -77,7 +77,7 @@ public void close() {
     }
 
     private void setup() {
-        final ProducerAppSetupConfiguration configuration = ProducerAppSetupConfiguration.builder()
+        final ProducerSetupConfiguration configuration = ProducerSetupConfiguration.builder()
                 .topics(this.topics)
                 .kafkaProperties(this.kafkaProperties)
                 .build();

streams-bootstrap/src/main/java/com/bakdata/kafka/ExecutableStreamsApp.java

@@ -66,6 +66,7 @@ public StreamsCleanUpRunner createCleanUpRunner() {
      */
     @Override
     public StreamsRunner createRunner() {
+        this.setup.run();
         return new StreamsRunner(this.topology, this.config);
     }
 

streams-bootstrap/src/main/java/com/bakdata/kafka/ProducerApp.java

@@ -45,7 +45,7 @@ public interface ProducerApp extends AutoCloseable {
      * Setup Kafka resources, such as topics, before running this app
      * @param configuration provides all runtime application configurations
      */
-    default void setup(final ProducerAppSetupConfiguration configuration) {
+    default void setup(final ProducerSetupConfiguration configuration) {
         // do nothing by default
     }
 

streams-bootstrap/src/main/java/com/bakdata/kafka/ProducerAppConfiguration.java

@@ -28,6 +28,7 @@
 
 import java.util.Map;
 import lombok.Builder;
+import lombok.EqualsAndHashCode;
 import lombok.NonNull;
 import lombok.Value;
 
@@ -36,6 +37,7 @@
  */
 @Builder
 @Value
+@EqualsAndHashCode
 public class ProducerAppConfiguration
         implements Configuration<ProducerApp, ConfiguredProducerApp<? extends ProducerApp>> {
     @Builder.Default

streams-bootstrap/src/main/java/com/bakdata/kafka/ProducerAppSetupConfiguration.java → streams-bootstrap/src/main/java/com/bakdata/kafka/ProducerSetupConfiguration.java

@@ -29,16 +29,18 @@
 import com.bakdata.kafka.util.ImprovedAdminClient;
 import java.util.Map;
 import lombok.Builder;
+import lombok.EqualsAndHashCode;
 import lombok.NonNull;
 import lombok.Value;
 
 /**
  * Configuration for setting up a {@link ProducerApp}
- * @see ProducerApp#setup(ProducerAppSetupConfiguration)
+ * @see ProducerApp#setup(ProducerSetupConfiguration)
  */
 @Builder
 @Value
-public class ProducerAppSetupConfiguration {
+@EqualsAndHashCode
+public class ProducerSetupConfiguration {
     @Builder.Default
     @NonNull ProducerTopicConfig topics = ProducerTopicConfig.builder().build();
     @Builder.Default

streams-bootstrap/src/main/java/com/bakdata/kafka/ProducerTopicConfig.java

@@ -29,6 +29,7 @@
 import com.google.common.base.Preconditions;
 import java.util.Map;
 import lombok.Builder;
+import lombok.EqualsAndHashCode;
 import lombok.NonNull;
 import lombok.Value;
 
@@ -37,6 +38,7 @@
  */
 @Builder
 @Value
+@EqualsAndHashCode
 public class ProducerTopicConfig {
 
     String outputTopic;

streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsApp.java

@@ -54,7 +54,7 @@ public interface StreamsApp extends AutoCloseable {
      * Setup Kafka resources, such as topics, before running this app
      * @param configuration provides all runtime application configurations
      */
-    default void setup(final StreamsAppSetupConfiguration configuration) {
+    default void setup(final StreamsSetupConfiguration configuration) {
         // do nothing by default
     }
 

streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsAppConfiguration.java

@@ -28,6 +28,7 @@
 
 import java.util.Map;
 import lombok.Builder;
+import lombok.EqualsAndHashCode;
 import lombok.NonNull;
 import lombok.Value;
 
@@ -36,6 +37,7 @@
  */
 @Builder
 @Value
+@EqualsAndHashCode
 public class StreamsAppConfiguration implements Configuration<StreamsApp, ConfiguredStreamsApp<? extends StreamsApp>> {
     @Builder.Default
     @NonNull StreamsTopicConfig topics = StreamsTopicConfig.builder().build();

streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsAppSetupConfiguration.java → streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsSetupConfiguration.java

@@ -29,16 +29,18 @@
 import com.bakdata.kafka.util.ImprovedAdminClient;
 import java.util.Map;
 import lombok.Builder;
+import lombok.EqualsAndHashCode;
 import lombok.NonNull;
 import lombok.Value;
 
 /**
  * Configuration for setting up a {@link StreamsApp}
- * @see StreamsApp#setup(StreamsAppSetupConfiguration)
+ * @see StreamsApp#setup(StreamsSetupConfiguration)
  */
 @Builder
 @Value
-public class StreamsAppSetupConfiguration {
+@EqualsAndHashCode
+public class StreamsSetupConfiguration {
     @Builder.Default
     @NonNull StreamsTopicConfig topics = StreamsTopicConfig.builder().build();
     @Builder.Default

streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsTopicConfig.java

@@ -32,6 +32,7 @@
 import java.util.Map;
 import java.util.regex.Pattern;
 import lombok.Builder;
+import lombok.EqualsAndHashCode;
 import lombok.NonNull;
 import lombok.Value;
 
@@ -40,6 +41,7 @@
  */
 @Builder
 @Value
+@EqualsAndHashCode
 public class StreamsTopicConfig {
 
     @Builder.Default

streams-bootstrap/src/test/java/com/bakdata/kafka/ExecutableProducerAppTest.java

@@ -0,0 +1,125 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2024 bakdata
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package com.bakdata.kafka;
+
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.STRICT_STUBS)
+class ExecutableProducerAppTest {
+
+    @Mock
+    private Consumer<ProducerSetupConfiguration> setup;
+    @Mock
+    private Supplier<ProducerCleanUpConfiguration> setupCleanUp;
+
+    @Test
+    void shouldCallSetupWhenCreatingRunner() {
+        final ProducerTopicConfig topics = ProducerTopicConfig.builder()
+                .outputTopic("output")
+                .build();
+        final ProducerAppConfiguration configuration = ProducerAppConfiguration.builder()
+                .topics(topics)
+                .build();
+        final ConfiguredProducerApp<ProducerApp> configuredApp = configuration.configure(new TestProducer());
+        final KafkaEndpointConfig endpointConfig = KafkaEndpointConfig.builder()
+                .brokers("localhost:9092")
+                .build();
+        final ExecutableProducerApp<ProducerApp> executableApp = configuredApp.withEndpoint(endpointConfig);
+        final Map<String, Object> kafkaProperties = configuredApp.getKafkaProperties(endpointConfig);
+        executableApp.createRunner();
+        verify(this.setup).accept(ProducerSetupConfiguration.builder()
+                .kafkaProperties(kafkaProperties)
+                .topics(topics)
+                .build());
+    }
+
+    @Test
+    void shouldCallSetupWhenCreatingRunnerWithOptions() {
+        final ProducerTopicConfig topics = ProducerTopicConfig.builder()
+                .outputTopic("output")
+                .build();
+        final ProducerAppConfiguration configuration = ProducerAppConfiguration.builder()
+                .topics(topics)
+                .build();
+        final ConfiguredProducerApp<ProducerApp> configuredApp = configuration.configure(new TestProducer());
+        final KafkaEndpointConfig endpointConfig = KafkaEndpointConfig.builder()
+                .brokers("localhost:9092")
+                .build();
+        final ExecutableProducerApp<ProducerApp> executableApp = configuredApp.withEndpoint(endpointConfig);
+        final Map<String, Object> kafkaProperties = configuredApp.getKafkaProperties(endpointConfig);
+        executableApp.createRunner(ProducerExecutionOptions.builder().build());
+        verify(this.setup).accept(ProducerSetupConfiguration.builder()
+                .kafkaProperties(kafkaProperties)
+                .topics(topics)
+                .build());
+    }
+
+    @Test
+    void shouldCallSetupCleanUpWhenCreatingCleanUpRunner() {
+        final ProducerTopicConfig topics = ProducerTopicConfig.builder()
+                .outputTopic("output")
+                .build();
+        final ProducerAppConfiguration configuration = ProducerAppConfiguration.builder()
+                .topics(topics)
+                .build();
+        final ConfiguredProducerApp<ProducerApp> configuredApp = configuration.configure(new TestProducer());
+        final KafkaEndpointConfig endpointConfig = KafkaEndpointConfig.builder()
+                .brokers("localhost:9092")
+                .build();
+        final ExecutableProducerApp<ProducerApp> executableApp = configuredApp.withEndpoint(endpointConfig);
+        when(this.setupCleanUp.get()).thenReturn(new ProducerCleanUpConfiguration());
+        executableApp.createCleanUpRunner();
+    }
+
+    private class TestProducer implements ProducerApp {
+
+        @Override
+        public void setup(final ProducerSetupConfiguration configuration) {
+            ExecutableProducerAppTest.this.setup.accept(configuration);
+        }
+
+        @Override
+        public ProducerCleanUpConfiguration setupCleanUp() {
+            return ExecutableProducerAppTest.this.setupCleanUp.get();
+        }
+
+        @Override
+        public ProducerRunnable buildRunnable(final ProducerBuilder builder) {
+            return () -> {};
+        }
+    }
+}