Add error topic deletion to clean up of consumer producer

streams-bootstrap-core/src/main/java/com/bakdata/kafka/consumer/ConsumerCleanUpRunner.java

@@ -1,7 +1,7 @@
 /*
  * MIT License
  *
- * Copyright (c) 2025 bakdata
+ * Copyright (c) 2026 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
@@ -114,9 +114,19 @@ private AdminClientX createAdminClient() {
 
     @RequiredArgsConstructor
     private class Task {
-
         private final @NonNull AdminClientX adminClient;
 
+        private static <T> T runAdminFuture(final KafkaFuture<T> action, final String errorMessage) {
+            try {
+                return action.get();
+            } catch (final InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new CleanUpException(errorMessage, e);
+            } catch (final ExecutionException e) {
+                throw new CleanUpException(errorMessage, e);
+            }
+        }
+
         private void reset() {
             final ConsumerGroupClient groupClient = this.adminClient.consumerGroups()
                     .group(ConsumerCleanUpRunner.this.groupId);
@@ -146,17 +156,6 @@ private void reset() {
             ConsumerCleanUpRunner.this.cleanHooks.runResetHooks();
         }
 
-        private static <T> T runAdminFuture(final KafkaFuture<T> action, final String errorMessage) {
-            try {
-                return action.get();
-            } catch (final InterruptedException e) {
-                Thread.currentThread().interrupt();
-                throw new CleanUpException(errorMessage, e);
-            } catch (final ExecutionException e) {
-                throw new CleanUpException(errorMessage, e);
-            }
-        }
-
         private void clean() {
             this.deleteConsumerGroup();
             ConsumerCleanUpRunner.this.cleanHooks.runCleanHooks();

streams-bootstrap-core/src/main/java/com/bakdata/kafka/consumerproducer/ConsumerProducerApp.java

@@ -26,13 +26,11 @@
 
 import com.bakdata.kafka.App;
 import com.bakdata.kafka.AppConfiguration;
-import com.bakdata.kafka.streams.StreamsCleanUpConfiguration;
-import com.bakdata.kafka.streams.StreamsCleanUpRunner;
 
 /**
  * Application that defines how to produce or consume messages to and from Kafka and necessary configurations
  */
-public interface ConsumerProducerApp extends App<ConsumerProducerTopicConfig, StreamsCleanUpConfiguration> {
+public interface ConsumerProducerApp extends App<ConsumerProducerTopicConfig, ConsumerProducerCleanUpConfiguration> {
 
     /**
      * Create a runnable that consumes and produces Kafka messages
@@ -46,8 +44,8 @@ public interface ConsumerProducerApp extends App<ConsumerProducerTopicConfig, St
      * This must be set to a unique value for every application interacting with your Kafka cluster to ensure internal
      * state encapsulation. Could be set to: className-outputTopic.
      * <p>
-     * User may provide a unique group identifier via {@link ConsumerProducerAppConfiguration#getUniqueAppId()}.
-     * If that is the case, the returned group ID should match the provided one.
+     * User may provide a unique group identifier via {@link ConsumerProducerAppConfiguration#getUniqueAppId()}. If that
+     * is the case, the returned group ID should match the provided one.
      *
      * @param configuration provides runtime configuration
      * @return unique group identifier
@@ -58,13 +56,13 @@ default String getUniqueAppId(final ConsumerProducerAppConfiguration configurati
     }
 
     /**
-     * @return {@code StreamsCleanUpConfiguration}
-     * @see StreamsCleanUpRunner
+     * @return {@code ConsumerProducerCleanUpConfiguration}
+     * @see ConsumerProducerCleanUpRunner
      */
     @Override
-    default StreamsCleanUpConfiguration setupCleanUp(
+    default ConsumerProducerCleanUpConfiguration setupCleanUp(
             final AppConfiguration<ConsumerProducerTopicConfig> configuration) {
-        return new StreamsCleanUpConfiguration();
+        return new ConsumerProducerCleanUpConfiguration();
     }
 
     @Override

streams-bootstrap-core/src/main/java/com/bakdata/kafka/consumerproducer/ConsumerProducerCleanUpConfiguration.java

@@ -0,0 +1,87 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2026 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.consumerproducer;
+
+import com.bakdata.kafka.HasCleanHook;
+import com.bakdata.kafka.HasTopicHooks;
+import java.util.ArrayList;
+import java.util.Collection;
+import lombok.NonNull;
+
+/**
+ * Provides configuration options for {@link ConsumerProducerCleanUpRunner}
+ */
+public class ConsumerProducerCleanUpConfiguration implements HasTopicHooks<ConsumerProducerCleanUpConfiguration>,
+        HasCleanHook<ConsumerProducerCleanUpConfiguration>, AutoCloseable {
+    private final @NonNull Collection<TopicHook> topicHooks = new ArrayList<>();
+    private final @NonNull Collection<Runnable> cleanHooks = new ArrayList<>();
+    private final @NonNull Collection<Runnable> resetHooks = new ArrayList<>();
+
+    /**
+     * Register a hook that is executed whenever a topic has been deleted by the cleanup runner.
+     */
+    @Override
+    public ConsumerProducerCleanUpConfiguration registerTopicHook(final TopicHook hook) {
+        this.topicHooks.add(hook);
+        return this;
+    }
+
+    /**
+     * Register a hook that is executed after {@link ConsumerProducerCleanUpRunner#clean()} has finished
+     */
+    @Override
+    public ConsumerProducerCleanUpConfiguration registerCleanHook(final Runnable hook) {
+        this.cleanHooks.add(hook);
+        return this;
+    }
+
+    /**
+     * Register a hook that is executed after {@link ConsumerProducerCleanUpRunner#reset()} has finished
+     *
+     * @param hook factory to create hook from
+     * @return self for chaining
+     */
+    public ConsumerProducerCleanUpConfiguration registerResetHook(final Runnable hook) {
+        this.resetHooks.add(hook);
+        return this;
+    }
+
+    @Override
+    public void close() {
+        this.topicHooks.forEach(TopicHook::close);
+    }
+
+    void runTopicDeletionHooks(final String topic) {
+        this.topicHooks.forEach(hook -> hook.deleted(topic));
+    }
+
+    void runCleanHooks() {
+        this.cleanHooks.forEach(Runnable::run);
+    }
+
+    void runResetHooks() {
+        this.resetHooks.forEach(Runnable::run);
+    }
+}

streams-bootstrap-core/src/main/java/com/bakdata/kafka/consumerproducer/ConsumerProducerCleanUpRunner.java

@@ -1,7 +1,7 @@
 /*
  * MIT License
  *
- * Copyright (c) 2025 bakdata
+ * Copyright (c) 2026 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
@@ -24,84 +24,172 @@
 
 package com.bakdata.kafka.consumerproducer;
 
+import com.bakdata.kafka.CleanUpException;
 import com.bakdata.kafka.CleanUpRunner;
-import com.bakdata.kafka.consumer.ConsumerCleanUpConfiguration;
-import com.bakdata.kafka.consumer.ConsumerCleanUpRunner;
+import com.bakdata.kafka.SchemaRegistryAppUtils;
+import com.bakdata.kafka.admin.AdminClientX;
+import com.bakdata.kafka.admin.ConsumerGroupsClient.ConsumerGroupClient;
 import com.bakdata.kafka.consumer.ConsumerTopicConfig;
-import com.bakdata.kafka.producer.ProducerCleanUpConfiguration;
-import com.bakdata.kafka.producer.ProducerCleanUpRunner;
-import com.bakdata.kafka.producer.ProducerTopicConfig;
-import com.bakdata.kafka.streams.StreamsCleanUpConfiguration;
 import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
 import lombok.AccessLevel;
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-
+import org.apache.kafka.clients.admin.ConsumerGroupDescription;
+import org.apache.kafka.clients.admin.ListOffsetsResult;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.GroupState;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.TopicPartition;
+import org.jooq.lambda.Seq;
 
 /**
+ * TODO:
+ * Clean up all topics specified by a {@link ConsumerTopicConfig}
  * Delete all output topics specified by a {@link ConsumerProducerTopicConfig}
  */
 @Slf4j
 @RequiredArgsConstructor(access = AccessLevel.PRIVATE)
 public final class ConsumerProducerCleanUpRunner implements CleanUpRunner {
-    private final @NonNull ConsumerCleanUpRunner consumerCleanUpRunner;
-    private final @NonNull ProducerCleanUpRunner producerCleanUpRunner;
+    private final @NonNull ConsumerProducerTopicConfig topics;
+    private final @NonNull Map<String, Object> kafkaProperties;
+    private final @NonNull String groupId;
+    private final @NonNull ConsumerProducerCleanUpConfiguration cleanHooks;
 
     /**
-     * Create a new {@code ConsumerProducerCleanUpRunner} with default {@link ConsumerCleanUpConfiguration}
+     * Create a new {@code ConsumerProducerCleanUpRunner} with default {@link ConsumerProducerCleanUpConfiguration}
      *
      * @param topics topic configuration
      * @param kafkaProperties configuration to connect to Kafka admin tools
+     * @param groupId consumer group id to clean up
      * @return {@code ConsumerProducerCleanUpRunner}
      */
     public static ConsumerProducerCleanUpRunner create(@NonNull final ConsumerProducerTopicConfig topics,
             @NonNull final Map<String, Object> kafkaProperties,
             @NonNull final String groupId) {
-        return create(topics, kafkaProperties, groupId, new StreamsCleanUpConfiguration());
+        return create(topics, kafkaProperties, groupId, new ConsumerProducerCleanUpConfiguration());
     }
 
     /**
      * Create a new {@code ConsumerProducerCleanUpRunner}
      *
      * @param topics topic configuration
      * @param kafkaProperties configuration to connect to Kafka admin tools
-     * @param groupId group id of the consumer
+     * @param groupId consumer group id to clean up
      * @param configuration configuration for hooks that are called when running {@link #clean()}
-     * @return {@code ConsumerCleanUpRunner}
+     * @return {@code ConsumerProducerCleanUpRunner}
      */
     public static ConsumerProducerCleanUpRunner create(@NonNull final ConsumerProducerTopicConfig topics,
             @NonNull final Map<String, Object> kafkaProperties,
             @NonNull final String groupId,
-            @NonNull final StreamsCleanUpConfiguration configuration) {
-        final ConsumerTopicConfig consumerTopicConfig = topics.toConsumerTopicConfig();
-        final ProducerTopicConfig producerTopicConfig = topics.toProducerTopicConfig();
-        final ConsumerCleanUpConfiguration consumerConfig = configuration.toConsumerCleanUpConfiguration();
-        final ProducerCleanUpConfiguration producerConfig = configuration.toProducerCleanUpConfiguration();
-        final ConsumerCleanUpRunner consumerCleanUpRunner =
-                ConsumerCleanUpRunner.create(consumerTopicConfig, kafkaProperties, groupId, consumerConfig);
-        final ProducerCleanUpRunner producerCleanUpRunner =
-                ProducerCleanUpRunner.create(producerTopicConfig, kafkaProperties, producerConfig);
-        return new ConsumerProducerCleanUpRunner(consumerCleanUpRunner, producerCleanUpRunner);
+            @NonNull final ConsumerProducerCleanUpConfiguration configuration) {
+        SchemaRegistryAppUtils.createTopicHook(kafkaProperties)
+                .ifPresent(configuration::registerTopicHook);
+        return new ConsumerProducerCleanUpRunner(topics, kafkaProperties, groupId, configuration);
     }
 
     @Override
     public void close() {
-        this.consumerCleanUpRunner.close();
-        this.producerCleanUpRunner.close();
+        this.cleanHooks.close();
     }
 
     @Override
     public void clean() {
-        this.consumerCleanUpRunner.clean();
-        this.producerCleanUpRunner.clean();
+        try (final AdminClientX adminClient = this.createAdminClient()) {
+            final Task task = new Task(adminClient);
+            task.clean();
+        }
     }
 
     /**
      * Reset your ConsumerProducer app by resetting consumer group offsets
      */
     public void reset() {
-        this.consumerCleanUpRunner.reset();
+        try (final AdminClientX adminClient = this.createAdminClient()) {
+            final Task task = new Task(adminClient);
+            task.reset();
+        }
+    }
+
+    private AdminClientX createAdminClient() {
+        return AdminClientX.create(this.kafkaProperties);
     }
 
+    @RequiredArgsConstructor
+    private class Task {
+        private final @NonNull AdminClientX adminClient;
+
+        private static <T> T runAdminFuture(final KafkaFuture<T> action, final String errorMessage) {
+            try {
+                return action.get();
+            } catch (final InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new CleanUpException(errorMessage, e);
+            } catch (final ExecutionException e) {
+                throw new CleanUpException(errorMessage, e);
+            }
+        }
+
+        private void clean() {
+            this.deleteConsumerGroup();
+            this.deleteTopics();
+            ConsumerProducerCleanUpRunner.this.cleanHooks.runCleanHooks();
+        }
+
+        private void reset() {
+            final ConsumerGroupClient groupClient = this.adminClient.consumerGroups()
+                    .group(ConsumerProducerCleanUpRunner.this.groupId);
+            final Optional<ConsumerGroupDescription> groupDescription = groupClient.describe();
+            if (groupDescription.isEmpty()) {
+                return;
+            }
+            if (groupDescription.get().groupState() != GroupState.EMPTY) {
+                throw new CleanUpException("Error resetting application, consumer group is not empty");
+            }
+
+            final Map<TopicPartition, OffsetAndMetadata> groupOffsets = groupClient.listOffsets();
+
+            final Map<TopicPartition, OffsetSpec> request = groupOffsets.keySet().stream()
+                    .collect(Collectors.toMap(tp -> tp, tp -> OffsetSpec.earliest()));
+            final Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> earliestOffsets =
+                    runAdminFuture(this.adminClient.admin().listOffsets(request).all(),
+                            "Error resetting application, beginning consumer group offset could not be found");
+
+            final Map<TopicPartition, OffsetAndMetadata> resetOffsets = earliestOffsets.entrySet().stream()
+                    .collect(Collectors.toMap(Map.Entry::getKey,
+                            e -> new OffsetAndMetadata(e.getValue().offset())));
+            runAdminFuture(
+                    this.adminClient.admin()
+                            .alterConsumerGroupOffsets(ConsumerProducerCleanUpRunner.this.groupId, resetOffsets)
+                            .all(), "Error resetting application, could not alter consumer group offsets");
+
+            ConsumerProducerCleanUpRunner.this.cleanHooks.runResetHooks();
+        }
+
+        private void deleteConsumerGroup() {
+            this.adminClient.consumerGroups().group(ConsumerProducerCleanUpRunner.this.groupId).deleteIfExists();
+        }
+
+        private void deleteTopics() {
+            final Iterable<String> outputTopics = this.getAllOutputTopics();
+            outputTopics.forEach(this::deleteTopic);
+        }
+
+        private void deleteTopic(final String topic) {
+            this.adminClient.topics()
+                    .topic(topic).deleteIfExists();
+            ConsumerProducerCleanUpRunner.this.cleanHooks.runTopicDeletionHooks(topic);
+        }
+
+        private Iterable<String> getAllOutputTopics() {
+            final String errorTopic = ConsumerProducerCleanUpRunner.this.topics.getErrorTopic();
+            return Seq.of(ConsumerProducerCleanUpRunner.this.topics.getOutputTopic())
+                    .concat(ConsumerProducerCleanUpRunner.this.topics.getLabeledOutputTopics().values())
+                    .append(Optional.ofNullable(errorTopic));
+        }
+    }
 }