Migrating Spring Kafka's SeekToCurrentErrorHandler to DefaultErrorHandler and updates method that was previously used to set the handler for AbstractKafkaListenerContainerFactory. (#900)

* Migrating Spring Kafka's `SeekToCurrentErrorHandler` to `DefaultErrorHandler` and updates method that was previously used to set the handler for `AbstractKafkaListenerContainerFactory`.

* Adding license that was missed and `@DocumentExample`

* Updating recipes.csv

* update type tables

* Helping a separate test to run on Windows

* Add version suffix when there are multiple spring-kafka entries

---------

Co-authored-by: David Grieve <david@moderne.io>
Co-authored-by: Tim te Beek <tim@moderne.io>

Author: 3 people

build.gradle.kts

@@ -123,6 +123,8 @@ recipeDependencies {
     testParserClasspath("org.springframework.batch:spring-batch-infrastructure:5.+")
     testParserClasspath("org.springframework.batch:spring-batch-test:4.3.+")
 
+    testParserClasspath("org.springframework.kafka:spring-kafka:2.8.+")
+
     testParserClasspath("org.springframework.boot:spring-boot-actuator:2.1.+")
     testParserClasspath("org.springframework.boot:spring-boot-actuator:3.0.+")
     testParserClasspath("org.springframework.boot:spring-boot-actuator:3.4.+")

src/main/resources/META-INF/rewrite/classpath.tsv.gz

@@ -42,3 +42,24 @@ recipeList:
   - org.openrewrite.java.ReplaceConstantWithAnotherConstant:
       existingFullyQualifiedConstantName: org.springframework.kafka.support.KafkaHeaders.RECEIVED_PARTITION_ID
       fullyQualifiedConstantName: org.springframework.kafka.support.KafkaHeaders.RECEIVED_PARTITION
+  - org.openrewrite.java.spring.kafka.UpgradeSpringKafka_2_8_ErrorHandlers
+---
+type: specs.openrewrite.org/v1beta/recipe
+name: org.openrewrite.java.spring.kafka.UpgradeSpringKafka_2_8_ErrorHandlers
+displayName: Migrates Spring Kafka deprecated error handlers
+description: >-
+  Migrate error handlers deprecated in Spring Kafka `2.8.x` to their replacements.
+tags:
+  - spring
+  - kafka
+recipeList:
+  # `BiConsumer<T>` arguments to `SeekToCurrentErrorHandler` constructor also change to `ConsumerRecordRecoverer`, but values are assignable already
+  - org.openrewrite.java.ChangeMethodName:
+      methodPattern: org.springframework.kafka.listener.SeekToCurrentErrorHandler handle(..)
+      newMethodName: handleRemaining
+  - org.openrewrite.java.ChangeMethodName:
+      methodPattern: org.springframework.kafka.config.AbstractKafkaListenerContainerFactory setErrorHandler(..)
+      newMethodName: setCommonErrorHandler
+  - org.openrewrite.java.ChangeType:
+      oldFullyQualifiedTypeName: org.springframework.kafka.listener.SeekToCurrentErrorHandler
+      newFullyQualifiedTypeName: org.springframework.kafka.listener.DefaultErrorHandler

src/main/resources/META-INF/rewrite/recipes.csv

@@ -34,7 +34,7 @@ public void defaults(RecipeSpec spec) {
             "spring-beans-5",
             "spring-context-5",
             "spring-core-5",
-            "spring-kafka-2",
+            "spring-kafka-2.9",
             "spring-messaging-5"
           ));
     }
@@ -51,7 +51,7 @@ void changeKafkaOperationsSendReturnType() {
               import org.springframework.kafka.support.SendResult;
               import org.springframework.util.concurrent.ListenableFuture;
               import org.springframework.util.concurrent.ListenableFutureCallback;
-              
+
               class Foo {
                   void bar(KafkaOperations<String, String> kafkaOperations) {
                       ListenableFuture<SendResult<String,String>> future = kafkaOperations.send("topic", "key", "value");
@@ -60,7 +60,7 @@ void bar(KafkaOperations<String, String> kafkaOperations) {
                           public void onSuccess(SendResult<String, String> result) {
                               System.out.println(result.getRecordMetadata());
                           }
-              
+
                           @Override
                           public void onFailure(Throwable ex) {
                               System.err.println(ex.getMessage());
@@ -72,9 +72,9 @@ public void onFailure(Throwable ex) {
             """
               import org.springframework.kafka.core.KafkaOperations;
               import org.springframework.kafka.support.SendResult;
-              
+
               import java.util.concurrent.CompletableFuture;
-              
+
               class Foo {
                   void bar(KafkaOperations<String, String> kafkaOperations) {
                       CompletableFuture<SendResult<String,String>> future = kafkaOperations.send("topic", "key", "value");
@@ -103,7 +103,7 @@ void changeKafkaTemplate() {
               import org.springframework.kafka.support.SendResult;
               import org.springframework.util.concurrent.ListenableFuture;
               import org.springframework.util.concurrent.ListenableFutureCallback;
-              
+
               class Foo {
                   void bar(KafkaTemplate<String, String> kafkaTemplate) {
                       ListenableFuture<SendResult<String,String>> future = kafkaTemplate.send("topic", "key", "value");
@@ -112,7 +112,7 @@ void bar(KafkaTemplate<String, String> kafkaTemplate) {
                           public void onSuccess(SendResult<String, String> result) {
                               System.out.println(result.getRecordMetadata());
                           }
-              
+
                           @Override
                           public void onFailure(Throwable ex) {
                               System.err.println(ex.getMessage());
@@ -124,9 +124,9 @@ public void onFailure(Throwable ex) {
             """
               import org.springframework.kafka.core.KafkaTemplate;
               import org.springframework.kafka.support.SendResult;
-              
+
               import java.util.concurrent.CompletableFuture;
-              
+
               class Foo {
                   void bar(KafkaTemplate<String, String> kafkaTemplate) {
                       CompletableFuture<SendResult<String,String>> future = kafkaTemplate.send("topic", "key", "value");
@@ -153,15 +153,15 @@ void noReplacementElsewhereYet() {
             """
               import org.springframework.util.concurrent.ListenableFuture;
               import org.springframework.util.concurrent.ListenableFutureCallback;
-              
+
               class Foo {
                   void bar(ListenableFuture<String> future) {
                       future.addCallback(new ListenableFutureCallback<>() {
                           @Override
                           public void onSuccess(String result) {
                               System.out.println(result);
                           }
-              
+
                           @Override
                           public void onFailure(Throwable ex) {
                               System.err.println(ex.getMessage());

src/main/resources/META-INF/rewrite/spring-kafka-30.yml

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2026 the original author or authors.
+ * <p>
+ * Licensed under the Moderne Source Available License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * https://docs.moderne.io/licensing/moderne-source-available-license
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.openrewrite.java.spring.kafka;
+
+import org.junit.jupiter.api.Test;
+import org.openrewrite.DocumentExample;
+import org.openrewrite.InMemoryExecutionContext;
+import org.openrewrite.java.JavaParser;
+import org.openrewrite.test.RecipeSpec;
+import org.openrewrite.test.RewriteTest;
+
+import static org.openrewrite.java.Assertions.java;
+
+class UpgradeSpringKafkaErrorHandlersTest implements RewriteTest {
+    @Override
+    public void defaults(RecipeSpec spec) {
+        spec
+          .recipeFromResources("org.openrewrite.java.spring.kafka.UpgradeSpringKafka_2_8_ErrorHandlers")
+          .parser(JavaParser.fromJavaVersion().classpathFromResources(new InMemoryExecutionContext(),
+            "kafka-clients",
+            "spring-beans",
+            "spring-context",
+            "spring-kafka-2.8"
+          ));
+    }
+
+    @DocumentExample
+    @Test
+    void migratesSeekToCurrentErrorHandler() {
+        rewriteRun(
+          //language=java
+          java(
+            """
+              import java.lang.Exception;
+              import java.util.List;
+
+              import org.apache.kafka.clients.consumer.Consumer;
+              import org.apache.kafka.clients.consumer.ConsumerRecord;
+              import org.springframework.kafka.config.AbstractKafkaListenerContainerFactory;
+              import org.springframework.kafka.listener.MessageListenerContainer;
+              import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
+
+              class A {
+                  private final SeekToCurrentErrorHandler handler = new SeekToCurrentErrorHandler();
+
+                  void method(AbstractKafkaListenerContainerFactory factory) {
+                      factory.setErrorHandler(handler);
+                  }
+
+                  void anotherMethod(
+                    Exception exception,
+                    List<ConsumerRecord<?,?>> records,
+                    Consumer<?,?> consumer,
+                    MessageListenerContainer container
+                  ) {
+                      handler.handle(exception, records, consumer, container);
+                  }
+              }
+              """,
+            """
+              import java.lang.Exception;
+              import java.util.List;
+
+              import org.apache.kafka.clients.consumer.Consumer;
+              import org.apache.kafka.clients.consumer.ConsumerRecord;
+              import org.springframework.kafka.config.AbstractKafkaListenerContainerFactory;
+              import org.springframework.kafka.listener.DefaultErrorHandler;
+              import org.springframework.kafka.listener.MessageListenerContainer;
+
+              class A {
+                  private final DefaultErrorHandler handler = new DefaultErrorHandler();
+
+                  void method(AbstractKafkaListenerContainerFactory factory) {
+                      factory.setCommonErrorHandler(handler);
+                  }
+
+                  void anotherMethod(
+                    Exception exception,
+                    List<ConsumerRecord<?,?>> records,
+                    Consumer<?,?> consumer,
+                    MessageListenerContainer container
+                  ) {
+                      handler.handleRemaining(exception, records, consumer, container);
+                  }
+              }
+              """
+          )
+        );
+    }
+}

src/test/java/org/openrewrite/java/spring/kafka/KafkaOperationsSendReturnTypeTest.java

@@ -24,6 +24,7 @@
 import org.openrewrite.test.RewriteTest;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.openrewrite.PathUtils.separatorsToSystem;
 import static org.openrewrite.java.Assertions.java;
 
 class FindSpringComponentsTest implements RewriteTest {
@@ -46,14 +47,14 @@ void findSpringComponents() {
                       SpringComponentRelationships.Row::getSourceFile,
                       SpringComponentRelationships.Row::getDependantType,
                       SpringComponentRelationships.Row::getDependencyType)
-                    .contains("test/Config.java", "test.A", "test.B"),
+                    .contains(separatorsToSystem("test/Config.java"), "test.A", "test.B"),
                 two ->
                   assertThat(two)
                     .extracting(
                       SpringComponentRelationships.Row::getSourceFile,
                       SpringComponentRelationships.Row::getDependantType,
                       SpringComponentRelationships.Row::getDependencyType)
-                    .contains("test/C.java", "test.C", "test.B"))),
+                    .contains(separatorsToSystem("test/C.java"), "test.C", "test.B"))),
           //language=java
           java("package test; public class B {}"),
           //language=java