Fix Sonarqube issues

bakdata · Aug 7, 2024 · 025ad1d · 025ad1d
1 parent 8b34869
commit 025ad1d
Show file tree

Hide file tree

Showing 5 changed files with 46 additions and 26 deletions.
diff --git a/streams-bootstrap-cli/src/main/java/com/bakdata/kafka/KafkaApplication.java b/streams-bootstrap-cli/src/main/java/com/bakdata/kafka/KafkaApplication.java
@@ -323,6 +323,8 @@ private interface Stoppable {
 
     /**
      * Provides access to a {@link CleanUpRunner} and closes the associated {@link ExecutableApp}
+     *
+     * @param <CR> type of {@link CleanUpRunner} used by this app
      */
     @RequiredArgsConstructor(access = AccessLevel.PROTECTED)
     public static class CleanableApp<CR extends CleanUpRunner> implements AutoCloseable, Stoppable {
@@ -349,6 +351,8 @@ public void stop() {
 
     /**
      * Provides access to a {@link Runner} and closes the associated {@link ExecutableApp}
+     *
+     * @param <R> type of {@link Runner} used by this app
      */
     @RequiredArgsConstructor(access = AccessLevel.PROTECTED)
     public static final class RunnableApp<R extends Runner> implements AutoCloseable, Stoppable {

diff --git a/streams-bootstrap-core/src/main/java/com/bakdata/kafka/util/ConsumerGroupClient.java b/streams-bootstrap-core/src/main/java/com/bakdata/kafka/util/ConsumerGroupClient.java
@@ -70,7 +70,7 @@ public void deleteConsumerGroup(final String groupName) {
             this.adminClient.deleteConsumerGroups(List.of(groupName))
                     .all()
                     .get(this.timeout.toSeconds(), TimeUnit.SECONDS);
-            log.info("Deleted consumer group'{}'", groupName);
+            log.info("Deleted consumer group '{}'", groupName);
         } catch (final InterruptedException ex) {
             Thread.currentThread().interrupt();
             throw new KafkaAdminException("Failed to delete consumer group " + groupName, ex);

diff --git a/streams-bootstrap-core/src/test/java/com/bakdata/kafka/ConfiguredProducerAppTest.java b/streams-bootstrap-core/src/test/java/com/bakdata/kafka/ConfiguredProducerAppTest.java
@@ -98,9 +98,10 @@ void shouldThrowIfKeySerializerHasBeenConfiguredDifferently() {
         ));
         final ConfiguredProducerApp<ProducerApp> configuredApp =
                 new ConfiguredProducerApp<>(new TestProducer(), configuration);
-        assertThatThrownBy(() -> configuredApp.getKafkaProperties(KafkaEndpointConfig.builder()
+        final KafkaEndpointConfig endpointConfig = KafkaEndpointConfig.builder()
                 .bootstrapServers("fake")
-                .build()))
+                .build();
+        assertThatThrownBy(() -> configuredApp.getKafkaProperties(endpointConfig))
                 .isInstanceOf(IllegalArgumentException.class)
                 .hasMessage("'key.serializer' should not be configured already");
     }
@@ -112,9 +113,10 @@ void shouldThrowIfValueSerializerHasBeenConfiguredDifferently() {
         ));
         final ConfiguredProducerApp<ProducerApp> configuredApp =
                 new ConfiguredProducerApp<>(new TestProducer(), configuration);
-        assertThatThrownBy(() -> configuredApp.getKafkaProperties(KafkaEndpointConfig.builder()
+        final KafkaEndpointConfig endpointConfig = KafkaEndpointConfig.builder()
                 .bootstrapServers("fake")
-                .build()))
+                .build();
+        assertThatThrownBy(() -> configuredApp.getKafkaProperties(endpointConfig))
                 .isInstanceOf(IllegalArgumentException.class)
                 .hasMessage("'value.serializer' should not be configured already");
     }
@@ -126,9 +128,10 @@ void shouldThrowIfBootstrapServersHasBeenConfiguredDifferently() {
         ));
         final ConfiguredProducerApp<ProducerApp> configuredApp =
                 new ConfiguredProducerApp<>(new TestProducer(), configuration);
-        assertThatThrownBy(() -> configuredApp.getKafkaProperties(KafkaEndpointConfig.builder()
+        final KafkaEndpointConfig endpointConfig = KafkaEndpointConfig.builder()
                 .bootstrapServers("fake")
-                .build()))
+                .build();
+        assertThatThrownBy(() -> configuredApp.getKafkaProperties(endpointConfig))
                 .isInstanceOf(IllegalArgumentException.class)
                 .hasMessage("'bootstrap.servers' should not be configured already");
     }
@@ -140,10 +143,11 @@ void shouldThrowIfSchemaRegistryHasBeenConfiguredDifferently() {
         ));
         final ConfiguredProducerApp<ProducerApp> configuredApp =
                 new ConfiguredProducerApp<>(new TestProducer(), configuration);
-        assertThatThrownBy(() -> configuredApp.getKafkaProperties(KafkaEndpointConfig.builder()
+        final KafkaEndpointConfig endpointConfig = KafkaEndpointConfig.builder()
                 .bootstrapServers("fake")
                 .schemaRegistryUrl("fake")
-                .build()))
+                .build();
+        assertThatThrownBy(() -> configuredApp.getKafkaProperties(endpointConfig))
                 .isInstanceOf(IllegalArgumentException.class)
                 .hasMessage("'schema.registry.url' should not be configured already");
     }

diff --git a/streams-bootstrap-core/src/test/java/com/bakdata/kafka/ConfiguredStreamsAppTest.java b/streams-bootstrap-core/src/test/java/com/bakdata/kafka/ConfiguredStreamsAppTest.java
@@ -98,9 +98,10 @@ void shouldThrowIfKeySerdeHasBeenConfiguredDifferently() {
         ));
         final ConfiguredStreamsApp<StreamsApp> configuredApp =
                 new ConfiguredStreamsApp<>(new TestApplication(), configuration);
-        assertThatThrownBy(() -> configuredApp.getKafkaProperties(KafkaEndpointConfig.builder()
+        final KafkaEndpointConfig endpointConfig = KafkaEndpointConfig.builder()
                 .bootstrapServers("fake")
-                .build()))
+                .build();
+        assertThatThrownBy(() -> configuredApp.getKafkaProperties(endpointConfig))
                 .isInstanceOf(IllegalArgumentException.class)
                 .hasMessage("'default.key.serde' should not be configured already");
     }
@@ -112,9 +113,10 @@ void shouldThrowIfValueSerdeHasBeenConfiguredDifferently() {
         ));
         final ConfiguredStreamsApp<StreamsApp> configuredApp =
                 new ConfiguredStreamsApp<>(new TestApplication(), configuration);
-        assertThatThrownBy(() -> configuredApp.getKafkaProperties(KafkaEndpointConfig.builder()
+        final KafkaEndpointConfig endpointConfig = KafkaEndpointConfig.builder()
                 .bootstrapServers("fake")
-                .build()))
+                .build();
+        assertThatThrownBy(() -> configuredApp.getKafkaProperties(endpointConfig))
                 .isInstanceOf(IllegalArgumentException.class)
                 .hasMessage("'default.value.serde' should not be configured already");
     }
@@ -126,9 +128,10 @@ void shouldThrowIfAppIdHasBeenConfiguredDifferently() {
         ));
         final ConfiguredStreamsApp<StreamsApp> configuredApp =
                 new ConfiguredStreamsApp<>(new TestApplication(), configuration);
-        assertThatThrownBy(() -> configuredApp.getKafkaProperties(KafkaEndpointConfig.builder()
+        final KafkaEndpointConfig endpointConfig = KafkaEndpointConfig.builder()
                 .bootstrapServers("fake")
-                .build()))
+                .build();
+        assertThatThrownBy(() -> configuredApp.getKafkaProperties(endpointConfig))
                 .isInstanceOf(IllegalArgumentException.class)
                 .hasMessage("'application.id' should not be configured already");
     }
@@ -140,9 +143,10 @@ void shouldThrowIfBootstrapServersHasBeenConfiguredDifferently() {
         ));
         final ConfiguredStreamsApp<StreamsApp> configuredApp =
                 new ConfiguredStreamsApp<>(new TestApplication(), configuration);
-        assertThatThrownBy(() -> configuredApp.getKafkaProperties(KafkaEndpointConfig.builder()
+        final KafkaEndpointConfig endpointConfig = KafkaEndpointConfig.builder()
                 .bootstrapServers("fake")
-                .build()))
+                .build();
+        assertThatThrownBy(() -> configuredApp.getKafkaProperties(endpointConfig))
                 .isInstanceOf(IllegalArgumentException.class)
                 .hasMessage("'bootstrap.servers' should not be configured already");
     }
@@ -154,10 +158,11 @@ void shouldThrowIfSchemaRegistryHasBeenConfiguredDifferently() {
         ));
         final ConfiguredStreamsApp<StreamsApp> configuredApp =
                 new ConfiguredStreamsApp<>(new TestApplication(), configuration);
-        assertThatThrownBy(() -> configuredApp.getKafkaProperties(KafkaEndpointConfig.builder()
+        final KafkaEndpointConfig endpointConfig = KafkaEndpointConfig.builder()
                 .bootstrapServers("fake")
                 .schemaRegistryUrl("fake")
-                .build()))
+                .build();
+        assertThatThrownBy(() -> configuredApp.getKafkaProperties(endpointConfig))
                 .isInstanceOf(IllegalArgumentException.class)
                 .hasMessage("'schema.registry.url' should not be configured already");
     }

diff --git a/streams-bootstrap-large-messages/src/main/java/com/bakdata/kafka/LargeMessageAppUtils.java b/streams-bootstrap-large-messages/src/main/java/com/bakdata/kafka/LargeMessageAppUtils.java
@@ -26,6 +26,8 @@
 
 import com.bakdata.kafka.HasTopicHooks.TopicHook;
 import java.util.Map;
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
 import lombok.experimental.UtilityClass;
 
 /**
@@ -44,13 +46,7 @@ public class LargeMessageAppUtils {
     public static TopicHook createTopicHook(final Map<String, Object> kafkaProperties) {
         final AbstractLargeMessageConfig largeMessageConfig = new AbstractLargeMessageConfig(kafkaProperties);
         final LargeMessageStoringClient storer = largeMessageConfig.getStorer();
-        //TODO: close storer once it implements AutoCloseable
-        return new TopicHook() {
-            @Override
-            public void deleted(final String topic) {
-                storer.deleteAllFiles(topic);
-            }
-        };
+        return new LargeMessageTopicHook(storer);
     }
 
     /**
@@ -78,4 +74,15 @@ public static <T> T registerTopicHook(
             final HasTopicHooks<T> cleanUpConfiguration, final EffectiveAppConfiguration<?> configuration) {
         return cleanUpConfiguration.registerTopicHook(createTopicHook(configuration));
     }
+
+    @RequiredArgsConstructor
+    private static class LargeMessageTopicHook implements TopicHook {
+        //TODO: close storer once it implements AutoCloseable
+        private final @NonNull LargeMessageStoringClient storer;
+
+        @Override
+        public void deleted(final String topic) {
+            this.storer.deleteAllFiles(topic);
+        }
+    }
 }