Move CLI options to Mixin classes (#383)

Author: raphala

streams-bootstrap-cli/src/main/java/com/bakdata/kafka/KafkaApplication.java

@@ -50,8 +50,6 @@
  * This class provides the following configuration options:
  * <ul>
  *     <li>{@link #bootstrapServers}</li>
- *     <li>{@link #outputTopic}</li>
- *     <li>{@link #labeledOutputTopics}</li>
  *     <li>{@link #schemaRegistryUrl}</li>
  *     <li>{@link #kafkaConfig}</li>
  * </ul>
@@ -82,11 +80,6 @@ public abstract class KafkaApplication<R extends Runner, CR extends CleanUpRunne
     // ConcurrentLinkedDeque required because calling #stop() causes asynchronous #run() calls to finish and thus
     // concurrently iterating and removing from #runners
     private final ConcurrentLinkedDeque<Stoppable> activeApps = new ConcurrentLinkedDeque<>();
-    @CommandLine.Option(names = "--output-topic", description = "Output topic")
-    private String outputTopic;
-    @CommandLine.Option(names = "--labeled-output-topics", split = ",",
-            description = "Additional labeled output topics")
-    private Map<String, String> labeledOutputTopics = emptyMap();
     @CommandLine.Option(names = {"--bootstrap-servers", "--bootstrap-server"}, required = true,
             description = "Kafka bootstrap servers to connect to")
     private String bootstrapServers;

streams-bootstrap-cli/src/main/java/com/bakdata/kafka/mixin/ErrorOptions.java

@@ -0,0 +1,41 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2025 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.mixin;
+
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import picocli.CommandLine;
+
+/**
+ * Shared CLI options to configure Kafka applications with an error topic.
+ */
+@Getter
+@Setter
+@NoArgsConstructor
+public class ErrorOptions {
+    @CommandLine.Option(names = "--error-topic", description = "Error topic")
+    private String errorTopic;
+}

streams-bootstrap-cli/src/main/java/com/bakdata/kafka/mixin/InputOptions.java

@@ -0,0 +1,57 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2025 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.mixin;
+
+import static java.util.Collections.emptyList;
+import static java.util.Collections.emptyMap;
+
+import com.bakdata.kafka.StringListConverter;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import picocli.CommandLine;
+import picocli.CommandLine.UseDefaultConverter;
+
+/**
+ * Shared CLI options to configure Kafka applications that consume input data.
+ */
+@Getter
+@Setter
+@NoArgsConstructor
+public class InputOptions {
+    @CommandLine.Option(names = "--input-topics", description = "Input topics", split = ",")
+    private List<String> inputTopics = emptyList();
+    @CommandLine.Option(names = "--input-pattern", description = "Input pattern")
+    private Pattern inputPattern;
+    @CommandLine.Option(names = "--labeled-input-topics", split = ",", description = "Additional labeled input topics",
+            converter = {UseDefaultConverter.class, StringListConverter.class})
+    private Map<String, List<String>> labeledInputTopics = emptyMap();
+    @CommandLine.Option(names = "--labeled-input-patterns", split = ",",
+            description = "Additional labeled input patterns")
+    private Map<String, Pattern> labeledInputPatterns = emptyMap();
+}

streams-bootstrap-cli/src/main/java/com/bakdata/kafka/mixin/OutputOptions.java

@@ -0,0 +1,47 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2025 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.mixin;
+
+import static java.util.Collections.emptyMap;
+
+import java.util.Map;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import picocli.CommandLine;
+
+/**
+ * Shared CLI options to configure Kafka applications that produce output data.
+ */
+@Getter
+@Setter
+@NoArgsConstructor
+public class OutputOptions {
+    @CommandLine.Option(names = "--output-topic", description = "Output topic")
+    private String outputTopic;
+    @CommandLine.Option(names = "--labeled-output-topics", split = ",",
+            description = "Additional labeled output topics")
+    private Map<String, String> labeledOutputTopics = emptyMap();
+}

streams-bootstrap-cli/src/main/java/com/bakdata/kafka/producer/KafkaProducerApplication.java

@@ -25,18 +25,25 @@
 package com.bakdata.kafka.producer;
 
 import com.bakdata.kafka.KafkaApplication;
+import com.bakdata.kafka.mixin.OutputOptions;
 import java.util.Optional;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import lombok.Setter;
 import lombok.ToString;
+import lombok.experimental.Delegate;
 import lombok.extern.slf4j.Slf4j;
 import picocli.CommandLine.Command;
+import picocli.CommandLine.Mixin;
 
 
 /**
  * <p>The base class for creating Kafka Producer applications.</p>
- * This class provides all configuration options provided by {@link KafkaApplication}.
+ * This class provides the following configuration options in addition to those provided by {@link KafkaApplication}:
+ * <ul>
+ *     <li>{@link #getOutputTopic()}</li>
+ *     <li>{@link #getLabeledOutputTopics()}</li>
+ * </ul>
  * To implement your Kafka Producer application inherit from this class and add your custom options.  Run it by
  * creating an instance of your class and calling {@link #startApplication(String[])} from your main.
  *
@@ -51,6 +58,9 @@
 public abstract class KafkaProducerApplication<T extends ProducerApp> extends
         KafkaApplication<ProducerRunner, ProducerCleanUpRunner, ProducerExecutionOptions, ExecutableProducerApp<T>,
                 ConfiguredProducerApp<T>, ProducerTopicConfig, T, ProducerAppConfiguration> {
+    @Mixin
+    @Delegate
+    private OutputOptions outputOptions = new OutputOptions();
 
     /**
      * Delete all output topics associated with the Kafka Producer application.

streams-bootstrap-cli/src/main/java/com/bakdata/kafka/streams/KafkaStreamsApplication.java

@@ -24,38 +24,36 @@
 
 package com.bakdata.kafka.streams;
 
-import static java.util.Collections.emptyList;
-import static java.util.Collections.emptyMap;
-
 import com.bakdata.kafka.KafkaApplication;
-import com.bakdata.kafka.StringListConverter;
-import java.util.List;
-import java.util.Map;
+import com.bakdata.kafka.mixin.ErrorOptions;
+import com.bakdata.kafka.mixin.InputOptions;
+import com.bakdata.kafka.mixin.OutputOptions;
 import java.util.Optional;
-import java.util.regex.Pattern;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import lombok.Setter;
 import lombok.ToString;
+import lombok.experimental.Delegate;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.streams.KafkaStreams;
 import org.apache.kafka.streams.KafkaStreams.StateListener;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
 import picocli.CommandLine;
 import picocli.CommandLine.Command;
-import picocli.CommandLine.UseDefaultConverter;
-
+import picocli.CommandLine.Mixin;
 
 /**
  * <p>The base class for creating Kafka Streams applications.</p>
  * This class provides the following configuration options in addition to those provided by {@link KafkaApplication}:
  * <ul>
- *     <li>{@link #inputTopics}</li>
- *     <li>{@link #inputPattern}</li>
- *     <li>{@link #errorTopic}</li>
- *     <li>{@link #labeledInputTopics}</li>
- *     <li>{@link #labeledInputPatterns}</li>
+ *     <li>{@link #getInputTopics()}</li>
+ *     <li>{@link #getInputPattern()}</li>
+ *     <li>{@link #getLabeledInputTopics()}</li>
+ *     <li>{@link #getLabeledInputPatterns()}</li>
+ *     <li>{@link #getOutputTopic()}</li>
+ *     <li>{@link #getLabeledOutputTopics()}</li>
+ *     <li>{@link #getErrorTopic()}</li>
  *     <li>{@link #volatileGroupInstanceId}</li>
  * </ul>
  * To implement your Kafka Streams application inherit from this class and add your custom options.  Run it by
@@ -72,32 +70,29 @@
 public abstract class KafkaStreamsApplication<T extends StreamsApp> extends
         KafkaApplication<StreamsRunner, StreamsCleanUpRunner, StreamsExecutionOptions,
                 ExecutableStreamsApp<T>, ConfiguredStreamsApp<T>, StreamsTopicConfig, T, StreamsAppConfiguration> {
-    @CommandLine.Option(names = "--input-topics", description = "Input topics", split = ",")
-    private List<String> inputTopics = emptyList();
-    @CommandLine.Option(names = "--input-pattern", description = "Input pattern")
-    private Pattern inputPattern;
-    @CommandLine.Option(names = "--error-topic", description = "Error topic")
-    private String errorTopic;
-    @CommandLine.Option(names = "--labeled-input-topics", split = ",", description = "Additional labeled input topics",
-            converter = {UseDefaultConverter.class, StringListConverter.class})
-    private Map<String, List<String>> labeledInputTopics = emptyMap();
-    @CommandLine.Option(names = "--labeled-input-patterns", split = ",",
-            description = "Additional labeled input patterns")
-    private Map<String, Pattern> labeledInputPatterns = emptyMap();
+    @Mixin
+    @Delegate
+    private InputOptions inputOptions = new InputOptions();
+    @Mixin
+    @Delegate
+    private OutputOptions outputOptions = new OutputOptions();
+    @Mixin
+    @Delegate
+    private ErrorOptions errorOptions = new ErrorOptions();
     @CommandLine.Option(names = "--volatile-group-instance-id", arity = "0..1",
             description = "Whether the group instance id is volatile, i.e., it will change on a Streams shutdown.")
     private boolean volatileGroupInstanceId;
     @CommandLine.Option(names = "--application-id",
             description = "Unique application ID to use for Kafka Streams. Can also be provided by implementing "
-                          + "StreamsApp#getUniqueAppId()")
+                    + "StreamsApp#getUniqueAppId()")
     private String applicationId;
 
     /**
      * Reset the Kafka Streams application. Additionally, delete the consumer group and all output and intermediate
      * topics associated with the Kafka Streams application.
      */
     @Command(description = "Reset the Kafka Streams application. Additionally, delete the consumer group and all "
-                           + "output and intermediate topics associated with the Kafka Streams application.")
+            + "output and intermediate topics associated with the Kafka Streams application.")
     @Override
     public void clean() {
         super.clean();
@@ -108,7 +103,7 @@ public void clean() {
      * application.
      */
     @Command(description = "Clear all state stores, consumer group offsets, and internal topics associated with the "
-                           + "Kafka Streams application.")
+            + "Kafka Streams application.")
     public void reset() {
         this.prepareClean();
         try (final CleanableApp<StreamsCleanUpRunner> app = this.createCleanableApp()) {
@@ -131,13 +126,13 @@ public final Optional<StreamsExecutionOptions> createExecutionOptions() {
     @Override
     public final StreamsTopicConfig createTopicConfig() {
         return StreamsTopicConfig.builder()
-                .inputTopics(this.inputTopics)
-                .labeledInputTopics(this.labeledInputTopics)
-                .inputPattern(this.inputPattern)
-                .labeledInputPatterns(this.labeledInputPatterns)
+                .inputTopics(this.getInputTopics())
+                .labeledInputTopics(this.getLabeledInputTopics())
+                .inputPattern(this.getInputPattern())
+                .labeledInputPatterns(this.getLabeledInputPatterns())
                 .outputTopic(this.getOutputTopic())
                 .labeledOutputTopics(this.getLabeledOutputTopics())
-                .errorTopic(this.errorTopic)
+                .errorTopic(this.getErrorTopic())
                 .build();
     }