feat(pom): Update Quarkus to 3.15.1

Related dependencies have been updated as well like restassured and kafka. Protobuf as well. Fixes #113 Moving to quarkiverse-parent version 18, impacts on formatting of Java files.
quarkiverse · Oct 11, 2024 · 32bd545 · 32bd545
1 parent 5eab6d5
commit 32bd545
Show file tree

Hide file tree

Showing 17 changed files with 155 additions and 151 deletions.
diff --git a/bom/test/pom.xml b/bom/test/pom.xml
@@ -9,7 +9,7 @@
   <artifactId>quarkus-kafka-streams-processor-test-bom</artifactId>
   <packaging>pom</packaging>
   <properties>
-    <kafka.version>3.8.0</kafka.version>
+    <kafka.version>3.7.1</kafka.version>
     <org.mock-server.version>5.15.0</org.mock-server.version>
   </properties>
   <dependencyManagement>
@@ -43,12 +43,12 @@
       <dependency>
         <groupId>io.rest-assured</groupId>
         <artifactId>json-path</artifactId>
-        <version>5.4.0</version>
+        <version>5.5.0</version>
       </dependency>
       <dependency>
         <groupId>org.springframework.kafka</groupId>
         <artifactId>spring-kafka-test</artifactId>
-        <version>3.1.3</version>
+        <version>3.2.4</version>
         <exclusions>
           <exclusion>
             <groupId>org.apache.kafka</groupId>

diff --git a/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/TopologyProducerTest.java b/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/TopologyProducerTest.java
@@ -115,7 +115,8 @@ private TopologyProducer newTopologyProducer(
         when(dlqConfig.topic()).thenReturn(Optional.ofNullable(dlq));
         when(sourceToTopicsMappingBuilder.sourceToTopicsMapping()).thenReturn(sourceToTopicMapping);
         when(sinkToTopicMappingBuilder.sinkToTopicMapping()).thenReturn(sinkToTopicMapping);
-        TopologyProducer topologyProducer = new TopologyProducer(kStreamsProcessorConfig, configCustomizer, sourceToTopicsMappingBuilder, sinkToTopicMappingBuilder, interceptors);
+        TopologyProducer topologyProducer = new TopologyProducer(kStreamsProcessorConfig, configCustomizer,
+                sourceToTopicsMappingBuilder, sinkToTopicMappingBuilder, interceptors);
         return topologyProducer;
     }
 

diff --git a/...erse/kafkastreamsprocessor/impl/configuration/DefaultTopologySerdesConfigurationTest.java b/...erse/kafkastreamsprocessor/impl/configuration/DefaultTopologySerdesConfigurationTest.java
@@ -55,7 +55,7 @@ void setup() {
         customizer = new DefaultTopologySerdesConfiguration(objectMapper);
     }
 
-  @Test
+    @Test
     void shouldSetIntrospectionSerializerByDefault() {
         when(configuration.getProcessorPayloadType()).thenReturn((Class) JSonPojo.class);
         when(configuration.getSinkValueSerializer()).thenReturn(null);
@@ -74,16 +74,16 @@ void shouldSetCustomizedSinkSerializerAndValueSerde() {
         assertThat(configuration.getSourceValueSerde(), is(equalTo(customSerde)));
     }
 
-  @Test
+    @Test
     void shouldSetProtobufSerdeWhenProcessorIsProtobuf() {
         when(configuration.getProcessorPayloadType()).thenReturn((Class) PingMessage.Ping.class);
         customizer.apply(configuration);
         verify(configuration).setSourceValueSerde(isA(KafkaProtobufSerde.class));
     }
 
-  @Test
+    @Test
     void shouldSetJacksonSerdeWhenProcessorIsAPojo() {
-        when(configuration.getProcessorPayloadType()).thenReturn((Class)JSonPojo.class);
+        when(configuration.getProcessorPayloadType()).thenReturn((Class) JSonPojo.class);
         customizer.apply(configuration);
         verify(configuration).setSourceValueSerde(isA(JacksonSerde.class));
     }

diff --git a/...iverse/kafkastreamsprocessor/impl/decorator/processor/CdiRequestContextDecoratorTest.java b/...iverse/kafkastreamsprocessor/impl/decorator/processor/CdiRequestContextDecoratorTest.java
@@ -52,11 +52,11 @@ class CdiRequestContextDecoratorTest {
     @Mock
     ManagedContext requestContext;
 
-  @BeforeEach
-  public void setup() {
-    when(container.requestContext()).thenReturn(requestContext);
-    decorator = new CdiRequestContextDecorator<>(processor, container);
-  }
+    @BeforeEach
+    public void setup() {
+        when(container.requestContext()).thenReturn(requestContext);
+        decorator = new CdiRequestContextDecorator<>(processor, container);
+    }
 
     @Test
     public void shouldActivateDeactivateWhenProcessIsCalled() {
@@ -66,10 +66,10 @@ public void shouldActivateDeactivateWhenProcessIsCalled() {
         verify(requestContext).terminate();
     }
 
-  @Test
-  public void shouldNotActivateRequestScopeIfAlreadyActivated() {
-    when(requestContext.isActive()).thenReturn(true);
-    decorator.process(new Record<>("Hello", "World", 0L));
-    verify(requestContext, never()).activate();
-  }
+    @Test
+    public void shouldNotActivateRequestScopeIfAlreadyActivated() {
+        when(requestContext.isActive()).thenReturn(true);
+        decorator.process(new Record<>("Hello", "World", 0L));
+        verify(requestContext, never()).activate();
+    }
 }
diff --git a/...streamsprocessor/impl/decorator/processor/PunctuatorDecorationProcessorDecoratorTest.java b/...streamsprocessor/impl/decorator/processor/PunctuatorDecorationProcessorDecoratorTest.java
@@ -53,7 +53,7 @@ class PunctuatorDecorationProcessorDecoratorTest {
     @Mock
     InternalProcessorContext<String, PingMessage.Ping> context;
 
-  @Test
+    @Test
     void contextWrapped() {
         when(decoratedPunctuators.get()).thenReturn(decoratedPunctuator);
         Processor<String, PingMessage.Ping, String, PingMessage.Ping> processor = new Processor<>() {
@@ -68,14 +68,14 @@ public void process(Record<String, PingMessage.Ping> record) {
             }
         };
 
-    PunctuatorDecorationProcessorDecorator<String, PingMessage.Ping, String, PingMessage.Ping> decorator = new PunctuatorDecorationProcessorDecorator<>(
-            processor,
-            decoratedPunctuators);
-  decorator.init(context);
-  decorator.process(new Record<>("blabla",PingMessage.Ping.newBuilder().setMessage("blabla").build(),0L,null));
-  decorator.close();
+        PunctuatorDecorationProcessorDecorator<String, PingMessage.Ping, String, PingMessage.Ping> decorator = new PunctuatorDecorationProcessorDecorator<>(
+                processor,
+                decoratedPunctuators);
+        decorator.init(context);
+        decorator.process(new Record<>("blabla", PingMessage.Ping.newBuilder().setMessage("blabla").build(), 0L, null));
+        decorator.close();
 
-    verify(context).schedule(Duration.ofSeconds(10), PunctuationType.WALL_CLOCK_TIME, decoratedPunctuator);
+        verify(context).schedule(Duration.ofSeconds(10), PunctuationType.WALL_CLOCK_TIME, decoratedPunctuator);
         verify(decoratedPunctuator).setRealPunctuatorInstance(punctuator);
     }
 

diff --git a/...iverse/kafkastreamsprocessor/impl/errors/LogAndSendToDlqExceptionHandlerDelegateTest.java b/...iverse/kafkastreamsprocessor/impl/errors/LogAndSendToDlqExceptionHandlerDelegateTest.java
@@ -97,37 +97,37 @@ class LogAndSendToDlqExceptionHandlerDelegateTest {
 
     @BeforeEach
     void setUp() {
-      when(kStreamsProcessorConfig.dlq()).thenReturn(dlqConfig);
+        when(kStreamsProcessorConfig.dlq()).thenReturn(dlqConfig);
     }
 
-  @Test
-  void shouldNotBlockAndSendToDlqIfPossible() {
-    when(record.key()).thenReturn(RECORD_KEY_BYTES);
-    when(record.value()).thenReturn(RECORD_VALUE_BYTES);
-    when(record.timestamp()).thenReturn(RECORD_TIMESTAMP);
-    when(record.topic()).thenReturn(INPUT_TOPIC);
-    when(record.partition()).thenReturn(PARTITION);
-    when(record.headers()).thenReturn(headers);
-    when(dlqConfig.topic()).thenReturn(Optional.of(DLQ_TOPIC));
-    when(kStreamsProcessorConfig.errorStrategy()).thenReturn(ErrorHandlingStrategy.DEAD_LETTER_QUEUE);
-    RecordHeaders headersWithMetadata = new RecordHeaders();
-    when(metadataHandler.withMetadata(any(Headers.class), anyString(), anyInt(), any(Exception.class)))
-        .thenReturn(headersWithMetadata);
-    when(kafkaClientSupplier.getProducer(any())).thenReturn(dlqProducerMock);
-
-    handler = new LogAndSendToDlqExceptionHandlerDelegate(kafkaClientSupplier, metrics, metadataHandler,
-      kStreamsProcessorConfig);
-    handler.configure(Collections.emptyMap());
-
-    DeserializationHandlerResponse response = handler.handle(context, record, exception);
-    assertEquals(DeserializationHandlerResponse.CONTINUE, response);
-
-    verify(dlqProducerMock)
-        .send(eq(new ProducerRecord<>(DLQ_TOPIC, null, RECORD_TIMESTAMP, RECORD_KEY_BYTES, RECORD_VALUE_BYTES,
-            headersWithMetadata)), any(Callback.class));
-    verify(metadataHandler).withMetadata(eq(headers), eq(INPUT_TOPIC), eq(PARTITION), eq(exception));
-    assertThat(metrics.processorErrorCounter().count(), closeTo(1d, 0.01d));
-  }
+    @Test
+    void shouldNotBlockAndSendToDlqIfPossible() {
+        when(record.key()).thenReturn(RECORD_KEY_BYTES);
+        when(record.value()).thenReturn(RECORD_VALUE_BYTES);
+        when(record.timestamp()).thenReturn(RECORD_TIMESTAMP);
+        when(record.topic()).thenReturn(INPUT_TOPIC);
+        when(record.partition()).thenReturn(PARTITION);
+        when(record.headers()).thenReturn(headers);
+        when(dlqConfig.topic()).thenReturn(Optional.of(DLQ_TOPIC));
+        when(kStreamsProcessorConfig.errorStrategy()).thenReturn(ErrorHandlingStrategy.DEAD_LETTER_QUEUE);
+        RecordHeaders headersWithMetadata = new RecordHeaders();
+        when(metadataHandler.withMetadata(any(Headers.class), anyString(), anyInt(), any(Exception.class)))
+                .thenReturn(headersWithMetadata);
+        when(kafkaClientSupplier.getProducer(any())).thenReturn(dlqProducerMock);
+
+        handler = new LogAndSendToDlqExceptionHandlerDelegate(kafkaClientSupplier, metrics, metadataHandler,
+                kStreamsProcessorConfig);
+        handler.configure(Collections.emptyMap());
+
+        DeserializationHandlerResponse response = handler.handle(context, record, exception);
+        assertEquals(DeserializationHandlerResponse.CONTINUE, response);
+
+        verify(dlqProducerMock)
+                .send(eq(new ProducerRecord<>(DLQ_TOPIC, null, RECORD_TIMESTAMP, RECORD_KEY_BYTES, RECORD_VALUE_BYTES,
+                        headersWithMetadata)), any(Callback.class));
+        verify(metadataHandler).withMetadata(eq(headers), eq(INPUT_TOPIC), eq(PARTITION), eq(exception));
+        assertThat(metrics.processorErrorCounter().count(), closeTo(1d, 0.01d));
+    }
 
     @Test
     void shouldOnlyContinueIfDefaultErrorStrategy() {
@@ -136,7 +136,7 @@ void shouldOnlyContinueIfDefaultErrorStrategy() {
         when(kStreamsProcessorConfig.dlq()).thenReturn(dlqConfig);
         when(dlqConfig.topic()).thenReturn(Optional.of(DLQ_TOPIC));
         handler = new LogAndSendToDlqExceptionHandlerDelegate(kafkaClientSupplier, metrics, metadataHandler,
-          kStreamsProcessorConfig);
+                kStreamsProcessorConfig);
         handler.configure(Collections.emptyMap());
 
         DeserializationHandlerResponse response = handler.handle(context, record, exception);
@@ -152,7 +152,7 @@ void shouldFailFastIfDlqStrategyWithoutTopic() {
         when(kStreamsProcessorConfig.dlq()).thenReturn(dlqConfig);
         when(kStreamsProcessorConfig.errorStrategy()).thenReturn(ErrorHandlingStrategy.DEAD_LETTER_QUEUE);
         handler = new LogAndSendToDlqExceptionHandlerDelegate(kafkaClientSupplier, metrics, metadataHandler,
-          kStreamsProcessorConfig);
+                kStreamsProcessorConfig);
 
         assertThrows(IllegalStateException.class, () -> handler.configure(Collections.emptyMap()));
     }

diff --git a/...ava/io/quarkiverse/kafkastreamsprocessor/impl/health/RebalancingTopicHealthCheckTest.java b/...ava/io/quarkiverse/kafkastreamsprocessor/impl/health/RebalancingTopicHealthCheckTest.java
@@ -45,13 +45,13 @@ public void setUp() {
         healthCheck.kafkaStreams = streams;
     }
 
-  @Test
-  public void testStateRunningHealthOK() {
-    when(streams.state()).thenReturn(KafkaStreams.State.RUNNING);
-    HealthCheckResponse response = healthCheck.call();
-    assertThat(response.getStatus(), equalTo(HealthCheckResponse.Status.UP));
-    assertThat(response.getData().get().get("state"), equalTo("RUNNING"));
-  }
+    @Test
+    public void testStateRunningHealthOK() {
+        when(streams.state()).thenReturn(KafkaStreams.State.RUNNING);
+        HealthCheckResponse response = healthCheck.call();
+        assertThat(response.getStatus(), equalTo(HealthCheckResponse.Status.UP));
+        assertThat(response.getData().get().get("state"), equalTo("RUNNING"));
+    }
 
     @Test
     public void testUndefinedStateHealthKO() {
@@ -60,13 +60,13 @@ public void testUndefinedStateHealthKO() {
         assertThat(response.getData().get().get("state"), nullValue());
     }
 
-  @Test
-  public void testStateRebalancingHealthKO() {
-    when(streams.state()).thenReturn(KafkaStreams.State.REBALANCING);
-    HealthCheckResponse response = healthCheck.call();
-    assertThat(response.getStatus(), equalTo(HealthCheckResponse.Status.DOWN));
-    assertThat(response.getData().get().get("state"), equalTo("REBALANCING"));
-  }
+    @Test
+    public void testStateRebalancingHealthKO() {
+        when(streams.state()).thenReturn(KafkaStreams.State.REBALANCING);
+        HealthCheckResponse response = healthCheck.call();
+        assertThat(response.getStatus(), equalTo(HealthCheckResponse.Status.DOWN));
+        assertThat(response.getData().get().get("state"), equalTo("REBALANCING"));
+    }
 
     @Test
     public void testStateKOIfKakfaStreamsNotInjected() {

diff --git a/...rse/kafkastreamsprocessor/sample/kafkatorest/PingClientProcessorQuarkusWithRetryTest.java b/...rse/kafkastreamsprocessor/sample/kafkatorest/PingClientProcessorQuarkusWithRetryTest.java
@@ -90,24 +90,26 @@ public void tearDown() {
         consumer.close(Durations.ONE_SECOND);
     }
 
-  @Test
-  void singleMessageWithRetry() {
-    // response "PONG"
-    when(client.ping())
-        .thenThrow(mock(RetryableException.class))
-        .thenThrow(mock(RetryableException.class))
-        .thenThrow(mock(RetryableException.class))
-        .thenReturn("PONG");
-
-    producer.send(new ProducerRecord<>(kStreamsProcessorConfig.input().topic().get(), Ping.newBuilder().setMessage("hello").build()));
-    producer.flush();
-
-    ConsumerRecord<String, Ping> singleRecord = KafkaTestUtils.getSingleRecord(consumer, kStreamsProcessorConfig.output().topic().get(),
-      Durations.TEN_SECONDS);
-    consumer.commitSync();
-
-    assertEquals("PONG of hello", singleRecord.value().getMessage());
-    verify(client, times(4)).ping(); // 3 retry + 1
-  }
+    @Test
+    void singleMessageWithRetry() {
+        // response "PONG"
+        when(client.ping())
+                .thenThrow(mock(RetryableException.class))
+                .thenThrow(mock(RetryableException.class))
+                .thenThrow(mock(RetryableException.class))
+                .thenReturn("PONG");
+
+        producer.send(new ProducerRecord<>(kStreamsProcessorConfig.input().topic().get(),
+                Ping.newBuilder().setMessage("hello").build()));
+        producer.flush();
+
+        ConsumerRecord<String, Ping> singleRecord = KafkaTestUtils.getSingleRecord(consumer,
+                kStreamsProcessorConfig.output().topic().get(),
+                Durations.TEN_SECONDS);
+        consumer.commitSync();
+
+        assertEquals("PONG of hello", singleRecord.value().getMessage());
+        verify(client, times(4)).ping(); // 3 retry + 1
+    }
 
 }
diff --git a/...astreamsprocessor/sample/kafkatorest/PingClientProcessorQuarkusWithoutRetryCatchTest.java b/...astreamsprocessor/sample/kafkatorest/PingClientProcessorQuarkusWithoutRetryCatchTest.java
@@ -86,18 +86,19 @@ public void tearDown() {
         consumer.close(Durations.ONE_SECOND);
     }
 
-  @Test
-  void singleMessageWithoutRetry() {
-    // throw Exception
-    when(client.ping())
-      .thenThrow(mock(RuntimeException.class));
-
-    producer.send(new ProducerRecord<>(kStreamsProcessorConfig.input().topic().get(), Ping.newBuilder().setMessage("hello").build()));
-    producer.flush();
-
-    assertThrows(IllegalStateException.class, () -> {
-      KafkaTestUtils.getSingleRecord(consumer, kStreamsProcessorConfig.output().topic().get(),
-        Durations.TEN_SECONDS);
-    });
-  }
+    @Test
+    void singleMessageWithoutRetry() {
+        // throw Exception
+        when(client.ping())
+                .thenThrow(mock(RuntimeException.class));
+
+        producer.send(new ProducerRecord<>(kStreamsProcessorConfig.input().topic().get(),
+                Ping.newBuilder().setMessage("hello").build()));
+        producer.flush();
+
+        assertThrows(IllegalStateException.class, () -> {
+            KafkaTestUtils.getSingleRecord(consumer, kStreamsProcessorConfig.output().topic().get(),
+                    Durations.TEN_SECONDS);
+        });
+    }
 }
diff --git a/...java/io/quarkiverse/kafkastreamsprocessor/sample/kafkatorest/PingClientProcessorTest.java b/...java/io/quarkiverse/kafkastreamsprocessor/sample/kafkatorest/PingClientProcessorTest.java
@@ -61,14 +61,14 @@ public void setup() {
         processor.init(context);
     }
 
-  @Test
-  public void replies_with_pong() {
-    when(client.ping()).thenReturn("world");
-    when(context.recordMetadata()).thenReturn(Optional.of(recordMetadata));
+    @Test
+    public void replies_with_pong() {
+        when(client.ping()).thenReturn("world");
+        when(context.recordMetadata()).thenReturn(Optional.of(recordMetadata));
 
-    processor.process(new Record<>("key", PingMessage.Ping.newBuilder().setMessage("hello").build(), 0L));
+        processor.process(new Record<>("key", PingMessage.Ping.newBuilder().setMessage("hello").build(), 0L));
 
-    verify(context).forward(captor.capture());
-    assertEquals("world of hello", captor.getValue().value().getMessage());
-  }
+        verify(context).forward(captor.capture());
+        assertEquals("world of hello", captor.getValue().value().getMessage());
+    }
 }
diff --git a/pom.xml b/pom.xml
@@ -4,7 +4,7 @@
   <parent>
     <groupId>io.quarkiverse</groupId>
     <artifactId>quarkiverse-parent</artifactId>
-    <version>16</version>
+    <version>18</version>
   </parent>
   <groupId>io.quarkiverse.kafkastreamsprocessor</groupId>
   <artifactId>quarkus-kafka-streams-processor-parent</artifactId>
@@ -32,9 +32,9 @@
     <maven.compiler.release>17</maven.compiler.release>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
-    <quarkus.version>3.10.0</quarkus.version>
+    <quarkus.version>3.15.1</quarkus.version>
 
-    <protobuf.version>3.24.1</protobuf.version>
+    <protobuf.version>3.25.5</protobuf.version>
   </properties>
 
   <repositories>