JAVA-32180 Upgrade spring-kafka-3 to boot-3 (#16120)

Author: anuragkumawat

spring-kafka-3/pom.xml

@@ -2,9 +2,9 @@
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <parent>
         <groupId>com.baeldung</groupId>
-        <artifactId>parent-boot-2</artifactId>
+        <artifactId>parent-boot-3</artifactId>
         <version>0.0.1-SNAPSHOT</version>
-        <relativePath>../parent-boot-2</relativePath>
+        <relativePath>../parent-boot-3</relativePath>
     </parent>
 
     <modelVersion>4.0.0</modelVersion>
@@ -22,6 +22,7 @@
         <dependency>
             <groupId>org.springframework.kafka</groupId>
             <artifactId>spring-kafka</artifactId>
+            <version>${spring-kafka.version}</version>
         </dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
@@ -54,8 +55,9 @@
 
     <properties>
         <java.version>17</java.version>
-        <kafka-version>3.0.12</kafka-version>
+        <spring-kafka.version>3.1.2</spring-kafka.version>
         <testcontainers.version>1.19.3</testcontainers.version>
         <awaitility.version>4.2.0</awaitility.version>
+        <start-class>org.springframework.boot.SpringApplication.Application</start-class>
     </properties>
 </project>

spring-kafka-3/src/main/java/com/baeldung/spring/kafka/deserialization/exception/KafkaErrorHandler.java

@@ -13,8 +13,9 @@ class KafkaErrorHandler implements CommonErrorHandler {
     private static final Logger log = LoggerFactory.getLogger(KafkaErrorHandler.class);
 
     @Override
-    public void handleRecord(Exception exception, ConsumerRecord<?, ?> record, Consumer<?, ?> consumer, MessageListenerContainer container) {
+    public boolean handleOne(Exception exception, ConsumerRecord<?, ?> record, Consumer<?, ?> consumer, MessageListenerContainer container) {
         handle(exception, consumer);
+        return true;
     }
 
     @Override

spring-kafka-3/src/main/java/com/baeldung/spring/kafka/viewheaders/KafkaMessageConsumer.java

@@ -20,13 +20,13 @@ public void listen(@Payload String message, @Headers Map<String, Object> headers
 
         String topicName = (String) headers.get(KafkaHeaders.TOPIC);
         System.out.println("Topic: " + topicName);
-        int partitionID = (int) headers.get(KafkaHeaders.RECEIVED_PARTITION_ID);
+        int partitionID = (int) headers.get(KafkaHeaders.RECEIVED_PARTITION);
         System.out.println("Partition ID: " + partitionID);
     }
 
     @KafkaListener(topics = { "my-topic" }, groupId = "my-consumer-group")
     public void listen(@Payload String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String topicName,
-        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
+        @Header(KafkaHeaders.RECEIVED_PARTITION) int partition) {
         System.out.println("Topic: " + topicName);
         System.out.println("Partition ID: " + partition);
     }