Remove deprecated version attribute and Bump CP 7.9.4

Author: loicgreffier

dsl/docker-compose.yml

@@ -1,8 +1,7 @@
 ---
-version: '2'
 services:
   broker:
-    image: confluentinc/cp-kafka:7.7.0
+    image: confluentinc/cp-kafka:7.9.4
     hostname: broker
     container_name: broker
     networks:
@@ -31,7 +30,7 @@ services:
       CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
 
   control-center:
-    image: confluentinc/cp-enterprise-control-center:7.7.0
+    image: confluentinc/cp-enterprise-control-center:7.9.4
     hostname: control-center
     container_name: control-center
     networks:
@@ -49,7 +48,7 @@ services:
       PORT: 9021
 
   init-kafka:
-    image: confluentinc/cp-kafka:7.7.0
+    image: confluentinc/cp-kafka:7.9.4
     hostname: init-kafka
     container_name: init-kafka
     networks:

dsl/src/main/java/com/michelin/kafka/error/handling/dsl/KafkaStreamsApp.java

@@ -47,19 +47,20 @@ public static void main(String[] args) {
         StreamsBuilder streamsBuilder = new StreamsBuilder();
         buildTopology(streamsBuilder);
 
-        try (KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(), properties)) {
-            kafkaStreams.setUncaughtExceptionHandler(
-                    exception -> StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT);
-            Runtime.getRuntime().addShutdownHook(new Thread(kafkaStreams::close));
+        KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(), properties);
 
-            kafkaStreams.start();
-        }
+        kafkaStreams.setUncaughtExceptionHandler(
+                _ -> StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT);
+
+        Runtime.getRuntime().addShutdownHook(new Thread(kafkaStreams::close));
+
+        kafkaStreams.start();
     }
 
     public static void buildTopology(StreamsBuilder streamsBuilder) {
         streamsBuilder.stream("delivery_booked_topic", Consumed.with(Serdes.String(), Serdes.String()))
                 .mapValues(KafkaStreamsApp::parseFromJson) // JsonSyntaxException
-                .filter((key, value) -> {
+                .filter((_, value) -> {
                     if (value.getNumberOfTires() < 0) {
                         throw new InvalidDeliveryException("Number of tires cannot be negative");
                     }

processor-api/docker-compose.yml

@@ -1,8 +1,7 @@
 ---
-version: '2'
 services:
   broker:
-    image: confluentinc/cp-kafka:7.7.0
+    image: confluentinc/cp-kafka:7.9.4
     hostname: broker
     container_name: broker
     networks:
@@ -31,7 +30,7 @@ services:
       CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
 
   control-center:
-    image: confluentinc/cp-enterprise-control-center:7.7.0
+    image: confluentinc/cp-enterprise-control-center:7.9.4
     hostname: control-center
     container_name: control-center
     networks:
@@ -49,7 +48,7 @@ services:
       PORT: 9021
 
   init-kafka:
-    image: confluentinc/cp-kafka:7.7.0
+    image: confluentinc/cp-kafka:7.9.4
     hostname: init-kafka
     container_name: init-kafka
     networks:

processor-api/src/main/java/com/michelin/kafka/error/handling/papi/CustomProcessor.java

@@ -50,7 +50,7 @@ public class CustomProcessor extends ContextualProcessor<String, DeliveryBooked,
     @Override
     public void init(ProcessorContext<String, String> context) {
         super.init(context);
-        context.schedule(Duration.ofMinutes(1), PunctuationType.WALL_CLOCK_TIME, timestamp -> {
+        context.schedule(Duration.ofMinutes(1), PunctuationType.WALL_CLOCK_TIME, _ -> {
             // Simulate an intentional exception
             throw new KaboomException("Simulated exception in punctuation");
         });

processor-api/src/main/java/com/michelin/kafka/error/handling/papi/KafkaStreamsApp.java

@@ -48,13 +48,13 @@ public static void main(String[] args) {
         StreamsBuilder streamsBuilder = new StreamsBuilder();
         buildTopology(streamsBuilder);
 
-        try (KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(), properties)) {
-            kafkaStreams.setUncaughtExceptionHandler(
-                    exception -> StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT);
-            Runtime.getRuntime().addShutdownHook(new Thread(kafkaStreams::close));
+        KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(), properties);
 
-            kafkaStreams.start();
-        }
+        kafkaStreams.setUncaughtExceptionHandler(
+                _ -> StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT);
+        Runtime.getRuntime().addShutdownHook(new Thread(kafkaStreams::close));
+
+        kafkaStreams.start();
     }
 
     public static void buildTopology(StreamsBuilder streamsBuilder) {