Fix name

eugenp · Dec 16, 2024 · 54a9ffc · 54a9ffc
1 parent e5b0852
commit 54a9ffc
Show file tree

Hide file tree

Showing 6 changed files with 9 additions and 17 deletions.
diff --git a/spring-reactive-modules/pom.xml b/spring-reactive-modules/pom.xml
@@ -31,6 +31,7 @@
         <module>spring-reactive-exceptions</module>
         <module>spring-reactor</module>
         <module>spring-webflux-amqp</module>
+        <module>spring-reactive-kafka-stream-binder</module>
 
         <!--    the following submodules are commented out as a workaround in order to use java 19+ and SB 3.2.x -->
         <!--    <module>spring-reactive-performance</module>-->

diff --git a/spring-reactive-modules/spring-reactive-kafka-stream-binder/pom.xml b/spring-reactive-modules/spring-reactive-kafka-stream-binder/pom.xml
@@ -145,9 +145,11 @@
     </build>
 
     <properties>
-        <spring-cloud.version>2023.0.2</spring-cloud.version>
+        <spring-cloud.version>2023.0.4</spring-cloud.version>
         <start-class>com.baeldung.webflux.kafka.stream.Application</start-class>
         <java.version>17</java.version>
+        <maven.compiler.release>17</maven.compiler.release>
+        <maven.compiler.release>17</maven.compiler.release>
     </properties>
 
 </project>
diff --git a/...am-binder/src/main/java/com/baeldung/reactive/kafka/stream/binder/domain/StockUpdate.java b/...am-binder/src/main/java/com/baeldung/reactive/kafka/stream/binder/domain/StockUpdate.java
@@ -1,12 +1,7 @@
 package com.baeldung.reactive.kafka.stream.binder.domain;
 
-import org.springframework.kafka.support.serializer.JsonSerde;
-
 import java.time.Instant;
 
 public record StockUpdate(String symbol, double price, String currency, Instant timestamp) {
 
-    // Required for Kafka serialization
-    public static class StockUpdateJsonSerde extends JsonSerde<StockUpdate> { }
-
 }
diff --git a/...ain/java/com/baeldung/reactive/kafka/stream/binder/kafka/producer/StockPriceProducer.java b/...ain/java/com/baeldung/reactive/kafka/stream/binder/kafka/producer/StockPriceProducer.java
@@ -22,7 +22,7 @@
 @Component
 public class StockPriceProducer {
 
-    public static final String[] stocks = {"AAPL", "GOOG", "MSFT", "AMZN", "TSLA"};
+    public static final String[] STOCKS = {"AAPL", "GOOG", "MSFT", "AMZN", "TSLA"};
     private static final String CURRENCY = "USD";
 
     private final ReactiveKafkaProducerTemplate<String, StockUpdate> kafkaProducer;
@@ -38,7 +38,7 @@ public StockPriceProducer(@NonNull KafkaProperties properties, @Qualifier(TopicC
     public Flux<SenderResult<Void>> produceStockPrices(int count) {
         return Flux.range(0, count)
                 .map(i -> {
-                    String stock = stocks[random.nextInt(stocks.length)];
+                    String stock = STOCKS[random.nextInt(STOCKS.length)];
                     double price = 100 + (200 * random.nextDouble());
                     return MessageBuilder.withPayload(new StockUpdate(stock, price, CURRENCY, Instant.now()))
                             .setHeader(MessageHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)

diff --git a/...g-reactive-modules/spring-reactive-kafka-stream-binder/src/main/resources/application.yml b/...g-reactive-modules/spring-reactive-kafka-stream-binder/src/main/resources/application.yml
@@ -5,8 +5,6 @@ spring:
       kafka:
         binder:
           brokers: ${KAFKA_BROKER}
-      function:
-        definition: processStockPrices
       bindings:
         default:
           content-type: application/json
@@ -18,11 +16,7 @@ spring:
           group: live-stock-consumers-y
           producer:
             useNativeEncoding: true
-            configuration:
-              key:
-                serializer: org.apache.kafka.common.serialization.StringSerializer
-              value:
-                serializer: com.baeldung.reactive.kafka.stream.binder.domain.StockUpdate$StockUpdateJsonSerde
+
   kafka:
     consumer:
       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer

diff --git a/...a/stream/binder/StockIntegrationTest.java → ...ve/kafka/stream/binder/StockLiveTest.java b/...a/stream/binder/StockIntegrationTest.java → ...ve/kafka/stream/binder/StockLiveTest.java
@@ -31,7 +31,7 @@
 
 @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
 @Testcontainers(parallel = true)
-class StockIntegrationTest {
+class StockLiveTest {
 
     @TestConfiguration
     public static class Configuration {
@@ -124,7 +124,7 @@ void shouldGetAllPersistedEvents() {
 
         updates.forEach(update -> {
             Assert.assertTrue(update.price() > 0);
-            Assert.assertTrue(Set.of(StockPriceProducer.stocks).contains(update.symbol()));
+            Assert.assertTrue(Set.of(StockPriceProducer.STOCKS).contains(update.symbol()));
             Assert.assertTrue(update.currency().equals("USD"));
             Assert.assertTrue(isBetween(update.timestamp(), start, end));
         });