bakdata · philipp94831 · Jul 19, 2024 · Apr 2, 2024 · Apr 2, 2024 · Apr 3, 2024
diff --git a/README.md b/README.md
@@ -26,7 +26,13 @@ You can add streams-bootstrap via Maven Central.
 #### Gradle
 
 ```gradle
-compile group: 'com.bakdata.kafka', name: 'streams-bootstrap', version: '2.1.1'
+implementation group: 'com.bakdata.kafka', name: 'streams-bootstrap-cli', version: '3.0.0'
+```
+
+With Kotlin DSL
+
+```gradle
+implementation(group = "com.bakdata.kafka", name = "streams-bootstrap-cli", version = "3.0.0")
 ```
 
 #### Maven
@@ -35,8 +41,8 @@ compile group: 'com.bakdata.kafka', name: 'streams-bootstrap', version: '2.1.1'
 
 <dependency>
     <groupId>com.bakdata.kafka</groupId>
-    <artifactId>streams-bootstrap</artifactId>
-    <version>2.1.1</version>
+  <artifactId>streams-bootstrap-cli</artifactId>
+  <version>3.0.0</version>
 </dependency>
 ```
 
@@ -52,8 +58,10 @@ and `getUniqueAppId()`. You can define the topology of your application in `buil
 
 ```java
 import com.bakdata.kafka.KafkaStreamsApplication;
-import java.util.Properties;
-import org.apache.kafka.streams.StreamsBuilder;
+import com.bakdata.kafka.StreamsApp;
+import com.bakdata.kafka.StreamsTopicConfig;
+import com.bakdata.kafka.TopologyBuilder;
+import java.util.Map;
 import org.apache.kafka.streams.kstream.KStream;
 
 public class StreamsBootstrapApplication extends KafkaStreamsApplication {
@@ -62,26 +70,30 @@ public class StreamsBootstrapApplication extends KafkaStreamsApplication {
     }
 
     @Override
-    public void buildTopology(final StreamsBuilder builder) {
-        final KStream<String, String> input =
-                builder.<String, String>stream(this.getInputTopics());
+    public StreamsApp createApp(final boolean cleanUp) {
+      return new StreamsApp() {
+        @Override
+        public void buildTopology(final TopologyBuilder builder) {
+          final KStream<String, String> input = builder.streamInput();
 
-        // your topology
+          // your topology
 
-        input.to(this.getOutputTopic());
-    }
-
-    @Override
-    public String getUniqueAppId() {
-        return "streams-bootstrap-app";
-    }
+          input.to(builder.getTopics().getOutputTopic());
+        }
 
-    // Optionally you can override the default streams bootstrap Kafka properties 
-    @Override
-    protected Properties createKafkaProperties() {
-        final Properties kafkaProperties = super.createKafkaProperties();
+        @Override
+        public String getUniqueAppId(final StreamsTopicConfig topics) {
+          return "streams-bootstrap-app-" + topics.getOutputTopic();
+        }
 
-        return kafkaProperties;
+        // Optionally you can define custom Kafka properties
+        @Override
+        public Map<String, Object> createKafkaProperties() {
+          return Map.of(
+                  // your config
+          );
+        }
+      };
     }
 }
 ```
@@ -92,6 +104,8 @@ The following configuration options are available:
 
 - `--schema-registry-url`: The URL of the Schema Registry
 
+- `--kafka-config`: Kafka Streams configuration (`<String=String>[,<String=String>...]`)
+
 - `--input-topics`: List of input topics (comma-separated)
 
 - `--input-pattern`: Pattern of input topics
@@ -100,8 +114,6 @@ The following configuration options are available:
 
 - `--error-topic`: A topic to write errors to
 
-- `--streams-config`: Kafka Streams configuration (`<String=String>[,<String=String>...]`)
-
 - `--extra-input-topics`: Additional named input topics if you need to specify multiple topics with different message
   types (`<String=String>[,<String=String>...]`)
 
@@ -113,41 +125,53 @@ The following configuration options are available:
 
 - `--volatile-group-instance-id`: Whether the group instance id is volatile, i.e., it will change on a Streams shutdown.
 
-- `--clean-up`: Whether the state of the Kafka Streams app, i.e., offsets and state stores and auto-created topics,
-  should be cleared instead of running the app
+- `--debug`: Configure logging to debug
+
+Additionally, the following commands are available:
 
-- `--delete-output`: Whether the output topics with their associated schemas and the consumer group should be deleted
-  during the cleanup
+- `clean`: Reset the Kafka Streams application. Additionally, delete the consumer group and all output and intermediate
+  topics associated with the Kafka Streams application.
 
-- `--debug`: Configure logging to debug
+- `reset`: Clear all state stores, consumer group offsets, and internal topics associated with the Kafka Streams
+  application.
 
 #### Kafka producer
 
 Create a subclass of `KafkaProducerApplication`.
 
 ```java
 import com.bakdata.kafka.KafkaProducerApplication;
-import java.util.Properties;
-import org.apache.kafka.clients.producer.KafkaProducer;
+import com.bakdata.kafka.ProducerApp;
+import com.bakdata.kafka.ProducerBuilder;
+import com.bakdata.kafka.ProducerRunnable;
+import java.util.Map;
+import org.apache.kafka.clients.producer.Producer;
 
 public class StreamsBootstrapApplication extends KafkaProducerApplication {
     public static void main(final String[] args) {
         startApplication(new StreamsBootstrapApplication(), args);
     }
 
     @Override
-    protected void runApplication() {
-        try (final KafkaProducer<Object, Object> producer = this.createProducer()) {
-            // your producer
+    public ProducerApp createApp(final boolean cleanUp) {
+      return new ProducerApp() {
+        @Override
+        public ProducerRunnable buildRunnable(final ProducerBuilder builder) {
+          return () -> {
+            try (final Producer<Object, Object> producer = builder.createProducer()) {
+              // your producer
+            }
+          };
         }
-    }
-
-    // Optionally you can override the default streams bootstrap Kafka properties 
-    @Override
-    protected Properties createKafkaProperties() {
-        final Properties kafkaProperties = super.createKafkaProperties();
 
-        return kafkaProperties;
+        // Optionally you can define custom Kafka properties
+        @Override
+        public Map<String, Object> createKafkaProperties() {
+          return Map.of(
+                  // your config
+          );
+        }
+      };
     }
 }
 ```
@@ -158,17 +182,18 @@ The following configuration options are available:
 
 - `--schema-registry-url`: The URL of the Schema Registry
 
-- `--output-topic`: The output topic
+- `--kafka-config`: Kafka producer configuration (`<String=String>[,<String=String>...]`)
 
-- `--streams-config`: Kafka producer configuration (`<String=String>[,<String=String>...]`)
+- `--output-topic`: The output topic
 
 - `--extra-output-topics`: Additional named output topics (`String=String>[,<String=String>...]`)
 
-- `--clean-up`: Whether the output topics and associated schemas of the producer app should be deleted instead of
-  running the app
-
 - `--debug`: Configure logging to debug
 
+Additionally, the following commands are available:
+
+- `clean`: Delete all output topics associated with the Kafka Producer application.
+
 ### Helm Charts
 
 For the configuration and deployment to Kubernetes, you can use

diff --git a/build.gradle.kts b/build.gradle.kts
@@ -1,7 +1,7 @@
 plugins {
     id("com.bakdata.release") version "1.4.0"
     id("com.bakdata.sonar") version "1.4.0"
-    id("com.bakdata.sonatype") version "1.4.0"
+    id("com.bakdata.sonatype") version "1.4.1"
     id("io.freefair.lombok") version "8.4"
 }
 
@@ -16,6 +16,7 @@ allprojects {
     repositories {
         mavenCentral()
         maven(url = "https://packages.confluent.io/maven/")
+        maven(url = "https://s01.oss.sonatype.org/content/repositories/snapshots")
     }
 }
 

diff --git a/charts/producer-app-cleanup-job/templates/job.yaml b/charts/producer-app-cleanup-job/templates/job.yaml
@@ -55,6 +55,8 @@ spec:
           imagePullPolicy: "{{ .Values.imagePullPolicy }}"
           resources:
 {{ toYaml .Values.resources | indent 12 }}
+          args:
+            - clean
           env:
             - name: ENV_PREFIX
               value: {{ .Values.configurationEnvPrefix }}_
@@ -74,8 +76,6 @@ spec:
             - name: "{{ .Values.configurationEnvPrefix }}_DEBUG"
               value: {{ .Values.debug | quote }}
             {{- end }}
-            - name: "{{ .Values.configurationEnvPrefix }}_CLEAN_UP"
-              value: "true"
             {{- if hasKey .Values.streams "outputTopic" }}
             - name: "{{ .Values.configurationEnvPrefix }}_OUTPUT_TOPIC"
               value: {{ .Values.streams.outputTopic | quote }}

diff --git a/charts/producer-app/README.md b/charts/producer-app/README.md
@@ -47,7 +47,7 @@ Alternatively, a YAML file that specifies the values for the parameters can be p
 ### Streams
 
 | Parameter                   | Description                                                                                                | Default |
-| --------------------------- | ---------------------------------------------------------------------------------------------------------- | ------- |
+|-----------------------------|------------------------------------------------------------------------------------------------------------|---------|
 | `streams.brokers`           | Comma separated list of Kafka brokers to connect to.                                                       |         |
 | `streams.schemaRegistryUrl` | URL of Schema Registry to connect to.                                                                      | `null`  |
 | `streams.config`            | Configurations for your [Kafka producer app](https://kafka.apache.org/documentation/#producerconfigs).     | `{}`    |

diff --git a/charts/streams-app-cleanup-job/templates/job.yaml b/charts/streams-app-cleanup-job/templates/job.yaml
@@ -55,6 +55,12 @@ spec:
           imagePullPolicy: "{{ .Values.imagePullPolicy }}"
           resources:
 {{ toYaml .Values.resources | indent 12 }}
+          args:
+            {{- if .Values.streams.deleteOutput }}
+            - clean
+            {{- else }}
+            - reset
+            {{- end }}
           env:
             - name: ENV_PREFIX
               value: {{ .Values.configurationEnvPrefix }}_
@@ -70,20 +76,10 @@ spec:
             - name: "{{ .Values.configurationEnvPrefix }}_SCHEMA_REGISTRY_URL"
               value: {{ .Values.streams.schemaRegistryUrl | quote }}
             {{- end }}
-            {{- if hasKey .Values.streams "productive" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_PRODUCTIVE"
-              value: {{ .Values.streams.productive | quote }}
-            {{- end }}
             {{- if hasKey .Values "debug" }}
             - name: "{{ .Values.configurationEnvPrefix }}_DEBUG"
               value: {{ .Values.debug | quote }}
             {{- end }}
-            - name: "{{ .Values.configurationEnvPrefix }}_CLEAN_UP"
-              value: "true"
-            {{- if hasKey .Values.streams "deleteOutput" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_DELETE_OUTPUT"
-              value: {{ .Values.streams.deleteOutput | quote }}
-            {{- end }}
             {{- if and (hasKey .Values.streams "inputTopics") (.Values.streams.inputTopics) }}
             - name: "{{ .Values.configurationEnvPrefix }}_INPUT_TOPICS"
               value: {{ .Values.streams.inputTopics | join "," | quote }}

diff --git a/charts/streams-app-cleanup-job/values.yaml b/charts/streams-app-cleanup-job/values.yaml
@@ -36,7 +36,6 @@ streams:
   extraOutputTopics: {}
 #   role: output
 #  errorTopic: error
-#  productive: true
   deleteOutput: false
 
 commandLine: {}

diff --git a/charts/streams-app/README.md b/charts/streams-app/README.md
@@ -50,21 +50,20 @@ Alternatively, a YAML file that specifies the values for the parameters can be p
 
 ### Streams
 
-| Parameter                            | Description                                                                                                                                                                      | Default |
-| ------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------- |
-| `streams.brokers`                    | Comma separated list of Kafka brokers to connect to.                                                                                                                             |         |
-| `streams.schemaRegistryUrl`          | URL of Schema Registry to connect to.                                                                                                                                            | `null`  |
-| `streams.staticMembership`           | Whether to use [Kafka Static Group Membership](https://cwiki.apache.org/confluence/display/KAFKA/KIP-345%3A+Introduce+static+membership+protocol+to+reduce+consumer+rebalances). | `false` |
-| `streams.optimizeLeaveGroupBehavior` | Enabling this optimizes the leave group behavior when a pod is terminated. Depends on the deployment kind, i.e., `statefulSet`. Requires the app to use streams-bootstrap 2.7+.  | `true`  |
-| `streams.config`                     | Configurations for your [Kafka Streams app](https://kafka.apache.org/documentation/#streamsconfigs).                                                                             | `{}`    |
-| `streams.inputTopics`                | List of input topics for your streams application.                                                                                                                               | `[]`    |
-| `streams.extraInputTopics`           | Map of additional named input topics if you need to specify multiple topics with different message types.                                                                        | `{}`    |
-| `streams.inputPattern`               | Input pattern of topics for your streams application.                                                                                                                            |         |
-| `streams.extraInputPatterns`         | Map of additional named input patterns if you need to specify multiple topics with different message types.                                                                      | `{}`    |
-| `streams.outputTopic`                | Output topic for your streams application.                                                                                                                                       |         |
-| `streams.extraOutputTopics`          | Map of additional named output topics if you need to specify multiple topics with different message types.                                                                       | `{}`    |
-| `streams.errorTopic`                 | Error topic for your streams application.                                                                                                                                        |         |
-| `streams.productive`                 | Whether to use Kafka configuration values that are more suitable for production environments.                                                                                    | `true`  |
+| Parameter                    | Description                                                                                                                                                                      | Default |
+|------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------|
+| `streams.brokers`            | Comma separated list of Kafka brokers to connect to.                                                                                                                             |         |
+| `streams.schemaRegistryUrl`  | URL of Schema Registry to connect to.                                                                                                                                            | `null`  |
+| `streams.staticMembership`   | Whether to use [Kafka Static Group Membership](https://cwiki.apache.org/confluence/display/KAFKA/KIP-345%3A+Introduce+static+membership+protocol+to+reduce+consumer+rebalances). | `false` |
+| `streams.config`             | Configurations for your [Kafka Streams app](https://kafka.apache.org/documentation/#streamsconfigs).                                                                             | `{}`    |
+| `streams.inputTopics`        | List of input topics for your streams application.                                                                                                                               | `[]`    |
+| `streams.extraInputTopics`   | Map of additional named input topics if you need to specify multiple topics with different message types.                                                                        | `{}`    |
+| `streams.inputPattern`       | Input pattern of topics for your streams application.                                                                                                                            |         |
+| `streams.extraInputPatterns` | Map of additional named input patterns if you need to specify multiple topics with different message types.                                                                      | `{}`    |
+| `streams.outputTopic`        | Output topic for your streams application.                                                                                                                                       |         |
+| `streams.extraOutputTopics`  | Map of additional named output topics if you need to specify multiple topics with different message types.                                                                       | `{}`    |
+| `streams.errorTopic`         | Error topic for your streams application.                                                                                                                                        |         |
+| `streams.productive`         | Whether to use Kafka configuration values that are more suitable for production environments.                                                                                    | `true`  |
 
 ### Other
 

diff --git a/charts/streams-app/templates/deployment.yaml b/charts/streams-app/templates/deployment.yaml
@@ -111,7 +111,7 @@ spec:
             - name: KAFKA_JMX_PORT
               value: "{{ .Values.jmx.port }}"
           {{- end }}
-          {{- if and (.Values.streams.optimizeLeaveGroupBehavior) (not .Values.statefulSet) }}
+          {{- if not .Values.statefulSet }}
             - name: "{{ .Values.configurationEnvPrefix }}_VOLATILE_GROUP_INSTANCE_ID"
               value: "true"
           {{- end }}
@@ -123,10 +123,6 @@ spec:
             - name: "{{ .Values.configurationEnvPrefix }}_SCHEMA_REGISTRY_URL"
               value: {{ .Values.streams.schemaRegistryUrl | quote }}
           {{- end }}
-          {{- if hasKey .Values.streams "productive" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_PRODUCTIVE"
-              value: {{ .Values.streams.productive | quote }}
-          {{- end }}
           {{- if hasKey .Values "debug" }}
             - name: "{{ .Values.configurationEnvPrefix }}_DEBUG"
               value: {{ .Values.debug | quote }}

diff --git a/charts/streams-app/values.yaml b/charts/streams-app/values.yaml
@@ -27,7 +27,6 @@ streams:
   # brokers: "test:9092"
   # schemaRegistryUrl: "url:1234"
   staticMembership: false
-  optimizeLeaveGroupBehavior: true
   config: {}
   #   max.poll.records: 500
   #   Note that YAML may convert large integers to scientific notation. Use Strings to avoid this.
@@ -46,7 +45,6 @@ streams:
   extraOutputTopics: {}
   #   role: output
   # errorTopic: error
-  # productive: true
 
 commandLine: {}
 #  MY_CLI_PARAM: "foo-bar"

diff --git a/gradle.properties b/gradle.properties
@@ -1,7 +1,13 @@
-version=2.23.1-SNAPSHOT
+version=3.0.0-SNAPSHOT
 org.gradle.caching=true
-org.gradle.parallel=true
+# running Kafka JUnit in parallel causes problems
+org.gradle.parallel=false
 kafkaVersion=3.6.1
+kafkaJunitVersion=3.6.0
 confluentVersion=7.6.0
-fluentKafkaVersion=2.13.1
-org.gradle.jvmargs=-Xmx2048m
+fluentKafkaVersion=2.14.0
+junitVersion=5.10.2
+mockitoVersion=5.11.0
+assertJVersion=3.25.3
+log4jVersion=2.23.1
+org.gradle.jvmargs=-Xmx4096m
diff --git a/settings.gradle b/settings.gradle
@@ -7,7 +7,8 @@ pluginManagement {
 rootProject.name = 'streams-bootstrap'
 
 include(
-        ":streams-bootstrap",
+        ":streams-bootstrap-core",
         ":streams-bootstrap-test",
         ":streams-bootstrap-large-messages",
+        ":streams-bootstrap-cli",
 )