Update

bakdata · Jun 25, 2024 · 566bbb8 · 566bbb8
1 parent 3cf0503
commit 566bbb8
Show file tree

Hide file tree

Showing 6 changed files with 20 additions and 8 deletions.
diff --git a/streams-bootstrap-cli/build.gradle.kts b/streams-bootstrap-cli/build.gradle.kts
@@ -12,9 +12,9 @@ dependencies {
     implementation(group = "org.apache.logging.log4j", name = "log4j-slf4j2-impl", version = log4jVersion)
 
     val junitVersion: String by project
+    testRuntimeOnly(group = "org.junit.jupiter", name = "junit-jupiter-engine", version = junitVersion)
     testImplementation(group = "org.junit.jupiter", name = "junit-jupiter-api", version = junitVersion)
     testImplementation(group = "org.junit.jupiter", name = "junit-jupiter-params", version = junitVersion)
-    testRuntimeOnly(group = "org.junit.jupiter", name = "junit-jupiter-engine", version = junitVersion)
     val assertJVersion: String by project
     testImplementation(group = "org.assertj", name = "assertj-core", version = assertJVersion)
     val mockitoVersion: String by project

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
@@ -327,6 +327,9 @@ private interface Stoppable {
         void stop();
     }
 
+    /**
+     * Provides access to a {@link CleanUpRunner} and closes the associated {@link ExecutableApp}
+     */
     @RequiredArgsConstructor
     public class CleanableApp implements AutoCloseable, Stoppable {
         private final @NonNull ExecutableApp<?, ?, ?> app;
@@ -339,12 +342,18 @@ public void close() {
             KafkaApplication.this.activeApps.remove(this);
         }
 
+        /**
+         * Close the app
+         */
         @Override
         public void stop() {
             this.app.close();
         }
     }
 
+    /**
+     * Provides access to a {@link Runner} and closes the associated {@link ExecutableApp}
+     */
     @RequiredArgsConstructor
     public class RunnableApp implements AutoCloseable, Stoppable {
         private final @NonNull ExecutableApp<?, ?, ?> app;
@@ -357,6 +366,9 @@ public void close() {
             KafkaApplication.this.activeApps.remove(this);
         }
 
+        /**
+         * Close the runner and app
+         */
         @Override
         public void stop() {
             this.runner.close();

diff --git a/streams-bootstrap-cli/src/test/java/com/bakdata/kafka/integration/RunProducerAppTest.java b/streams-bootstrap-cli/src/test/java/com/bakdata/kafka/integration/RunProducerAppTest.java
@@ -78,8 +78,8 @@ void shouldRunApp() throws InterruptedException {
             public ProducerRunnable buildRunnable(final ProducerBuilder builder) {
                 return () -> {
                     try (final Producer<String, TestRecord> producer = builder.createProducer()) {
-                        final TestRecord record = TestRecord.newBuilder().setContent("bar").build();
-                        producer.send(new ProducerRecord<>(builder.getTopics().getOutputTopic(), "foo", record));
+                        final TestRecord testRecord = TestRecord.newBuilder().setContent("bar").build();
+                        producer.send(new ProducerRecord<>(builder.getTopics().getOutputTopic(), "foo", testRecord));
                     }
                 };
             }

diff --git a/streams-bootstrap-core/build.gradle.kts b/streams-bootstrap-core/build.gradle.kts
@@ -22,10 +22,10 @@ dependencies {
     implementation(group = "org.jooq", name = "jool", version = "0.9.14")
 
     val junitVersion: String by project
+    testRuntimeOnly(group = "org.junit.jupiter", name = "junit-jupiter-engine", version = junitVersion)
     testImplementation(group = "org.junit.jupiter", name = "junit-jupiter-api", version = junitVersion)
     testImplementation(group = "org.junit.jupiter", name = "junit-jupiter-params", version = junitVersion)
     testImplementation(group = "org.junit-pioneer", name = "junit-pioneer", version = "2.2.0")
-    testRuntimeOnly(group = "org.junit.jupiter", name = "junit-jupiter-engine", version = junitVersion)
     val assertJVersion: String by project
     testImplementation(group = "org.assertj", name = "assertj-core", version = assertJVersion)
     val mockitoVersion: String by project

diff --git a/streams-bootstrap-core/src/main/java/com/bakdata/kafka/StreamsExecutionOptions.java b/streams-bootstrap-core/src/main/java/com/bakdata/kafka/StreamsExecutionOptions.java
@@ -47,7 +47,7 @@ public class StreamsExecutionOptions {
      * Hook that is called after calling {@link KafkaStreams#start()}
      */
     @Builder.Default
-    private final @NonNull Consumer<RunningStreams> onStart = (runningStreams) -> {};
+    private final @NonNull Consumer<RunningStreams> onStart = runningStreams -> {};
     /**
      * Configures {@link KafkaStreams#setStateListener(StateListener)}
      */

diff --git a/streams-bootstrap-core/src/test/java/com/bakdata/kafka/AvroMirrorTest.java b/streams-bootstrap-core/src/test/java/com/bakdata/kafka/AvroMirrorTest.java
@@ -49,18 +49,18 @@ private static ConfiguredStreamsApp<MirrorWithNonDefaultSerde> createApp() {
     void shouldMirror() {
         final Serde<TestRecord> keySerde = this.getKeySerde();
         final Serde<TestRecord> valueSerde = this.getValueSerde();
-        final TestRecord record = TestRecord.newBuilder()
+        final TestRecord testRecord = TestRecord.newBuilder()
                 .setContent("bar")
                 .build();
         this.testTopology.input()
                 .withKeySerde(keySerde)
                 .withValueSerde(valueSerde)
-                .add(record, record);
+                .add(testRecord, testRecord);
 
         this.testTopology.streamOutput()
                 .withKeySerde(keySerde)
                 .withValueSerde(valueSerde)
-                .expectNextRecord().hasKey(record).hasValue(record)
+                .expectNextRecord().hasKey(testRecord).hasValue(testRecord)
                 .expectNoMoreRecord();
     }