review

bakdata · Feb 16, 2024 · 590aec3 · 590aec3
1 parent 24a3092
commit 590aec3
Showing 1 changed file with 11 additions and 13 deletions.
diff --git a/src/main/java/com/bakdata/kafka/KafkaConnectorSourceResetter.java b/src/main/java/com/bakdata/kafka/KafkaConnectorSourceResetter.java
@@ -122,18 +122,6 @@ private static Converter createConverter(final Map<String, Object> kafkaConfig)
         return converter;
     }
 
-    private static <K, V> List<PartitionInfo> partitionsFor(final Consumer<K, V> consumer, final String topic) {
-        final Map<String, List<PartitionInfo>> topicsWithPartition = consumer.listTopics();
-        if (!topicsWithPartition.containsKey(topic)) {
-            final String message = String.format(
-                    "Could not fetch partitions from the offset topic '%s'. Check if the offset topic name is set "
-                            + "correctly.",
-                    topic);
-            throw new IllegalArgumentException(message);
-        }
-        return topicsWithPartition.get(topic);
-    }
-
     @Override
     public void run() {
         final String id = this.createId();
@@ -147,6 +135,7 @@ public void run() {
         log.info("Finished resetting {}", this.sharedOptions.getConnectorName());
     }
 
+
     private void resetPartitions(final Iterable<byte[]> partitions, final Map<String, Object> kafkaConfig) {
         try (final Producer<byte[], byte[]> producer = createProducer(kafkaConfig)) {
             producer.initTransactions();
@@ -191,7 +180,7 @@ private Consumer<byte[], byte[]> createConsumer(final Map<String, Object> kafkaC
         final Deserializer<byte[]> byteArrayDeserializer = new ByteArrayDeserializer();
         final Consumer<byte[], byte[]> consumer =
                 new KafkaConsumer<>(kafkaConfig, byteArrayDeserializer, byteArrayDeserializer);
-        final List<PartitionInfo> partitions = partitionsFor(consumer, this.offsetTopic);
+        final List<PartitionInfo> partitions = this.partitionsForOffsetTopic(consumer);
         final List<TopicPartition> topicPartitions = partitions.stream()
                 .map(KafkaConnectorSourceResetter::toTopicPartition)
                 .collect(Collectors.toList());
@@ -200,4 +189,13 @@ private Consumer<byte[], byte[]> createConsumer(final Map<String, Object> kafkaC
         return consumer;
     }
 
+    private <K, V> List<PartitionInfo> partitionsForOffsetTopic(final Consumer<K, V> consumer) {
+        final Map<String, List<PartitionInfo>> topicsWithPartition = consumer.listTopics();
+        if (!topicsWithPartition.containsKey(this.offsetTopic)) {
+            final String message = String.format("Topic %s does not exist.", this.offsetTopic);
+            throw new IllegalArgumentException(message);
+        }
+        return topicsWithPartition.get(this.offsetTopic);
+    }
+
 }