Create v3

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,46 +58,61 @@ 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.SerdeConfig;
+import com.bakdata.kafka.StreamsApp;
+import com.bakdata.kafka.StreamsTopicConfig;
+import com.bakdata.kafka.TopologyBuilder;
+import java.util.Map;
+import org.apache.kafka.common.serialization.Serdes.StringSerde;
 import org.apache.kafka.streams.kstream.KStream;
 
-public class StreamsBootstrapApplication extends KafkaStreamsApplication {
+public class MyStreamsApplication extends KafkaStreamsApplication {
     public static void main(final String[] args) {
-        startApplication(new StreamsBootstrapApplication(), args);
+      startApplication(new MyStreamsApplication(), args);
     }
 
     @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());
-    }
+          input.to(builder.getTopics().getOutputTopic());
+        }
 
-    @Override
-    public String getUniqueAppId() {
-        return "streams-bootstrap-app";
-    }
+        @Override
+        public String getUniqueAppId(final StreamsTopicConfig topics) {
+          return "streams-bootstrap-app-" + topics.getOutputTopic();
+        }
 
-    // Optionally you can override the default streams bootstrap Kafka properties 
-    @Override
-    protected Properties createKafkaProperties() {
-        final Properties kafkaProperties = super.createKafkaProperties();
+        @Override
+        public SerdeConfig defaultSerializationConfig() {
+          return new SerdeConfig(StringSerde.class, StringSerde.class);
+        }
 
-        return kafkaProperties;
+        // Optionally you can define custom Kafka properties
+        @Override
+        public Map<String, Object> createKafkaProperties() {
+          return Map.of(
+                  // your config
+          );
+        }
+      };
     }
 }
 ```
 
 The following configuration options are available:
 
-- `--brokers`: List of Kafka brokers (comma-separated) (**required**)
+- `--bootstrap-servers`, `--bootstrap-server`: List of Kafka bootstrap servers (comma-separated) (**required**)
 
 - `--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,74 +121,91 @@ 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>...]`)
+- `--labeled-input-topics`: Additional labeled input topics if you need to specify multiple topics with different
+  message types (`<String=String>[,<String=String>...]`)
 
-- `--extra-input-patterns`: Additional named input patterns if you need to specify multiple topics with different
+- `--labeled-input-patterns`: Additional labeled input patterns if you need to specify multiple topics with different
   message types (`<String=String>[,<String=String>...]`)
 
-- `--extra-output-topics`: Additional named output topics if you need to specify multiple topics with different message
-  types (`String=String>[,<String=String>...]`)
+- `--labeled-output-topics`: Additional labeled output topics if you need to specify multiple topics with different
+  message types (`String=String>[,<String=String>...]`)
+
+- `--application-id`: Unique application ID to use for Kafka Streams. Can also be provided by
+  implementing `StreamsApp#getUniqueAppId()`
 
 - `--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
+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;
-
-public class StreamsBootstrapApplication extends KafkaProducerApplication {
+import com.bakdata.kafka.ProducerApp;
+import com.bakdata.kafka.ProducerBuilder;
+import com.bakdata.kafka.ProducerRunnable;
+import com.bakdata.kafka.SerializerConfig;
+import java.util.Map;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.common.serialization.StringSerializer;
+
+public class MyProducerApplication extends KafkaProducerApplication {
     public static void main(final String[] args) {
-        startApplication(new StreamsBootstrapApplication(), args);
+      startApplication(new MyProducerApplication(), 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();
+        @Override
+        public SerializerConfig defaultSerializationConfig() {
+          return new SerializerConfig(StringSerializer.class, StringSerializer.class);
+        }
 
-        return kafkaProperties;
+        // Optionally you can define custom Kafka properties
+        @Override
+        public Map<String, Object> createKafkaProperties() {
+          return Map.of(
+                  // your config
+          );
+        }
+      };
     }
 }
 ```
 
 The following configuration options are available:
 
-- `--brokers`: List of Kafka brokers (comma-separated) (**required**)
+- `--bootstrap-servers`, `--bootstrap-server`: List of Kafka bootstrap servers (comma-separated) (**required**)
 
 - `--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>...]`)
+- `--labeled-output-topics`: Additional labeled 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
+Additionally, the following commands are available:
 
-- `--debug`: Configure logging to debug
+- `clean`: Delete all output topics associated with the Kafka Producer application.
 
 ### Helm Charts
 

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")
     }
 }
 

charts/producer-app-cleanup-job/templates/job.yaml

@@ -55,34 +55,30 @@ spec:
           imagePullPolicy: "{{ .Values.imagePullPolicy }}"
           resources:
 {{ toYaml .Values.resources | indent 12 }}
+          args:
+            - clean
           env:
             - name: ENV_PREFIX
               value: {{ .Values.configurationEnvPrefix }}_
-            {{- range $key, $value := .Values.streams.config }}
-            - name: {{ printf "STREAMS_%s" $key | replace "." "_" | upper | quote }}
+            {{- range $key, $value := .Values.kafka.config }}
+            - name: {{ printf "KAFKA_%s" $key | replace "." "_" | upper | quote }}
               value: {{ $value | quote }}
             {{- end }}
-            {{- if hasKey .Values.streams "brokers" }}
-            - name: "{{ .Values.configurationEnvPrefix }}_BROKERS"
-              value: {{ .Values.streams.brokers | quote }}
+            {{- if hasKey .Values.kafka "bootstrapServers" }}
+            - name: "{{ .Values.configurationEnvPrefix }}_BOOTSTRAP_SERVERS"
+              value: {{ .Values.kafka.bootstrapServers | quote }}
             {{- end }}
-            {{- if hasKey .Values.streams "schemaRegistryUrl" }}
+            {{- if hasKey .Values.kafka "schemaRegistryUrl" }}
             - name: "{{ .Values.configurationEnvPrefix }}_SCHEMA_REGISTRY_URL"
-              value: {{ .Values.streams.schemaRegistryUrl | quote }}
+              value: {{ .Values.kafka.schemaRegistryUrl | 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 "outputTopic" }}
+            {{- if hasKey .Values.kafka "outputTopic" }}
             - name: "{{ .Values.configurationEnvPrefix }}_OUTPUT_TOPIC"
-              value: {{ .Values.streams.outputTopic | quote }}
+              value: {{ .Values.kafka.outputTopic | quote }}
             {{- end }}
-            {{- if and (hasKey .Values.streams "extraOutputTopics") (.Values.streams.extraOutputTopics) }}
-            - name: "{{ .Values.configurationEnvPrefix }}_EXTRA_OUTPUT_TOPICS"
-              value: "{{- range $key, $value := .Values.streams.extraOutputTopics }}{{ $key }}={{ $value }},{{- end }}"
+            {{- if and (hasKey .Values.kafka "labeledOutputTopics") (.Values.kafka.labeledOutputTopics) }}
+            - name: "{{ .Values.configurationEnvPrefix }}_LABELED_OUTPUT_TOPICS"
+              value: "{{- range $key, $value := .Values.kafka.labeledOutputTopics }}{{ $key }}={{ $value }},{{- end }}"
             {{- end }}
             {{- range $key, $value := .Values.secrets }}
             - name: "{{ $key }}"

charts/producer-app-cleanup-job/values.yaml

@@ -15,22 +15,20 @@ files: {}
 #    mountPath: app/resources
 #    content: "foo bar"
 
-streams:
-#  brokers: "test:9092"
+kafka:
+#  bootstrapServers: "test:9092"
 #  schemaRegistryUrl: "url:1234"
   config: {}
 #    max.poll.records: 500
 #    Note that YAML may convert large integers to scientific notation. Use Strings to avoid this.
 #    max.request.size: "1000000"
 #  outputTopic: output
-  extraOutputTopics: {}
-#   role: output
+  labeledOutputTopics: {}
+#   label: output
 
 commandLine: {}
 #  MY_CLI_PARAM: "foo-bar"
 
-debug: false
-
 env: {}
 #  MY_ENV_VARIABLE: foo-bar