Merge branch 'spring-projects:main' into spring-projectsGH-3001

Author: notizklotz

.github/dependabot.yml

@@ -23,3 +23,14 @@ updates:
          - "org.awaitility:awaitility"
          - "com.google.code.findbugs:jsr305"
          - "org.springframework.boot*"
+
+  - package-ecosystem: "github-actions"
+    directory: "/"
+    schedule:
+      interval: "weekly"
+      day: "saturday"
+    labels: ["type: task"]
+    groups:
+      development-dependencies:
+        patterns:
+          - "*"

.github/workflows/auto-cherry-pick.yml

@@ -0,0 +1,13 @@
+name: Auto Cherry-Pick
+
+on:
+  push:
+    branches:
+      - main
+      - '*.x'
+
+jobs:
+  cherry-pick-commit:
+    uses: spring-io/spring-github-workflows/.github/workflows/spring-cherry-pick.yml@main
+    secrets:
+      GH_ACTIONS_REPO_TOKEN: ${{ secrets.GH_ACTIONS_REPO_TOKEN }}

.github/workflows/backport-issue.yml

@@ -7,6 +7,6 @@ on:
 
 jobs:
   backport-issue:
-    uses: spring-io/spring-github-workflows/.github/workflows/spring-backport-issue.yml@v2
+    uses: spring-io/spring-github-workflows/.github/workflows/spring-backport-issue.yml@main
     secrets:
       GH_ACTIONS_REPO_TOKEN: ${{ secrets.GH_ACTIONS_REPO_TOKEN }}

.github/workflows/ci-snapshot.yml

@@ -17,11 +17,12 @@ concurrency:
 
 jobs:
   build-snapshot:
-    uses: spring-io/spring-github-workflows/.github/workflows/spring-artifactory-gradle-snapshot.yml@v2
+    uses: spring-io/spring-github-workflows/.github/workflows/spring-artifactory-gradle-snapshot.yml@main
     with:
       gradleTasks: ${{ github.event_name == 'schedule' && '--rerun-tasks' || '' }}
     secrets:
       GRADLE_ENTERPRISE_CACHE_USER: ${{ secrets.GRADLE_ENTERPRISE_CACHE_USER }}
       GRADLE_ENTERPRISE_CACHE_PASSWORD: ${{ secrets.GRADLE_ENTERPRISE_CACHE_PASSWORD }}
       GRADLE_ENTERPRISE_SECRET_ACCESS_KEY: ${{ secrets.GRADLE_ENTERPRISE_SECRET_ACCESS_KEY }}
-      JF_ARTIFACTORY_SPRING: ${{ secrets.JF_ARTIFACTORY_SPRING }}
+      ARTIFACTORY_USERNAME: ${{ secrets.ARTIFACTORY_USERNAME }}
+      ARTIFACTORY_PASSWORD: ${{ secrets.ARTIFACTORY_PASSWORD }}

.github/workflows/deploy-docs.yml

@@ -16,4 +16,4 @@ permissions:
 jobs:
   dispatch-docs-build:
     if: github.repository_owner == 'spring-projects'
-    uses: spring-io/spring-github-workflows/.github/workflows/spring-dispatch-docs-build.yml@v2
+    uses: spring-io/spring-github-workflows/.github/workflows/spring-dispatch-docs-build.yml@main

.github/workflows/merge-dependabot-pr.yml

@@ -11,4 +11,6 @@ jobs:
   merge-dependabot-pr:
     permissions: write-all
 
-    uses: spring-io/spring-github-workflows/.github/workflows/spring-merge-dependabot-pr.yml@v2
+    uses: spring-io/spring-github-workflows/.github/workflows/spring-merge-dependabot-pr.yml@main
+    with:
+      mergeArguments: --auto --squash

.github/workflows/pr-build.yml

@@ -7,4 +7,4 @@ on:
 
 jobs:
   build-pull-request:
-    uses: spring-io/spring-github-workflows/.github/workflows/spring-gradle-pull-request-build.yml@v2
+    uses: spring-io/spring-github-workflows/.github/workflows/spring-gradle-pull-request-build.yml@main

.github/workflows/release.yml

@@ -12,13 +12,15 @@ jobs:
       contents: write
       issues: write
 
-    uses: spring-io/spring-github-workflows/.github/workflows/spring-artifactory-gradle-release.yml@v2
+    uses: spring-io/spring-github-workflows/.github/workflows/spring-artifactory-gradle-release.yml@main
     secrets:
       GH_ACTIONS_REPO_TOKEN: ${{ secrets.GH_ACTIONS_REPO_TOKEN }}
       GRADLE_ENTERPRISE_CACHE_USER: ${{ secrets.GRADLE_ENTERPRISE_CACHE_USER }}
       GRADLE_ENTERPRISE_CACHE_PASSWORD: ${{ secrets.GRADLE_ENTERPRISE_CACHE_PASSWORD }}
       GRADLE_ENTERPRISE_SECRET_ACCESS_KEY: ${{ secrets.GRADLE_ENTERPRISE_SECRET_ACCESS_KEY }}
       JF_ARTIFACTORY_SPRING: ${{ secrets.JF_ARTIFACTORY_SPRING }}
+      ARTIFACTORY_USERNAME: ${{ secrets.ARTIFACTORY_USERNAME }}
+      ARTIFACTORY_PASSWORD: ${{ secrets.ARTIFACTORY_PASSWORD }}
       OSSRH_URL: ${{ secrets.OSSRH_URL }}
       OSSRH_S01_TOKEN_USERNAME: ${{ secrets.OSSRH_S01_TOKEN_USERNAME }}
       OSSRH_S01_TOKEN_PASSWORD: ${{ secrets.OSSRH_S01_TOKEN_PASSWORD }}

.github/workflows/verify-staged-artifacts.yml

@@ -8,6 +8,13 @@ on:
         required: true
         type: string
 
+env:
+  GRADLE_ENTERPRISE_CACHE_USERNAME: ${{ secrets.GRADLE_ENTERPRISE_CACHE_USER }}
+  GRADLE_ENTERPRISE_CACHE_PASSWORD: ${{ secrets.GRADLE_ENTERPRISE_CACHE_PASSWORD }}
+  GRADLE_ENTERPRISE_ACCESS_KEY: ${{ secrets.GRADLE_ENTERPRISE_SECRET_ACCESS_KEY }}
+  ARTIFACTORY_USERNAME: ${{ secrets.ARTIFACTORY_USERNAME }}
+  ARTIFACTORY_PASSWORD: ${{ secrets.ARTIFACTORY_PASSWORD }}
+
 jobs:
   verify-staged-with-spring-integration:
     runs-on: ubuntu-latest
@@ -19,28 +26,31 @@ jobs:
           repository: spring-projects/spring-integration
           show-progress: false
 
-      - name: Set up JDK
-        uses: actions/setup-java@v3
-        with:
-          distribution: temurin
-          java-version: 17
-          cache: 'gradle'
-
-      - uses: jfrog/setup-jfrog-cli@v3
-        env:
-          JF_ENV_SPRING: ${{ secrets.JF_ARTIFACTORY_SPRING }}
-
-      - name: Configure JFrog Cli
-        run: jf gradlec --repo-resolve libs-release-staging
+      - name: Set up Gradle
+        uses: spring-io/spring-gradle-build-action@v2
 
-      - name: Verify Spring Integration Kafka against staged release
+      - name: Prepare Spring Integration project against Staging
         run: |
-          sed -i "1,/springKafkaVersion.*/s/springKafkaVersion.*/springKafkaVersion='${{ inputs.releaseVersion }}'/" build.gradle
-          jf gradle :spring-integration-kafka:check
+          printf "allprojects {
+                  repositories {
+                    maven { 
+                      url 'https://repo.spring.io/libs-staging-local'
+                      credentials {
+                        username = '$ARTIFACTORY_USERNAME'
+                        password = '$ARTIFACTORY_PASSWORD'
+                      }
+                    }
+                  }
+                }" > staging-repo-init.gradle
+          
+          sed -i "1,/springKafkaVersion.*/s/springKafkaVersion.*/springKafkaVersion='${{ inputs.releaseVersion }}'/" build.gradle  
+
+      - name: Verify Spring Integration Kafka module against staged release
+        run: gradle :spring-integration-kafka:check --init-script staging-repo-init.gradle
 
       - name: Capture Test Results
         if: failure()
-        uses: actions/upload-artifact@v3
+        uses: actions/upload-artifact@v4
         with:
           name: test-results
           path: '**/target/surefire-reports/**/*.*'

build.gradle

@@ -55,22 +55,23 @@ ext {
 	awaitilityVersion = '4.2.0'
 	hamcrestVersion = '2.2'
 	hibernateValidationVersion = '8.0.1.Final'
-	jacksonBomVersion = '2.15.3'
+	jacksonBomVersion = '2.15.4'
 	jaywayJsonPathVersion = '2.8.0'
 	junit4Version = '4.13.2'
-	junitJupiterVersion = '5.10.1'
+	junitJupiterVersion = '5.10.2'
 	kafkaVersion = '3.6.1'
+	kotlinCoroutinesVersion = '1.7.3'
 	log4jVersion = '2.22.1'
 	micrometerDocsVersion = '1.0.2'
-	micrometerVersion = '1.13.0-SNAPSHOT'
-	micrometerTracingVersion = '1.3.0-SNAPSHOT'
+	micrometerVersion = '1.13.0-M1'
+	micrometerTracingVersion = '1.3.0-M1'
 	mockitoVersion = '5.8.0'
-	reactorVersion = '2023.0.2'
+	reactorVersion = '2023.0.3'
 	scalaVersion = '2.13'
 	springBootVersion = '3.2.2' // docs module
-	springDataVersion = '2023.2.0-SNAPSHOT'
+	springDataVersion = '2024.0.0-M1'
 	springRetryVersion = '2.0.5'
-	springVersion = '6.1.3'
+	springVersion = '6.1.4'
 	zookeeperVersion = '3.8.3'
 
 	idPrefix = 'kafka'
@@ -278,6 +279,7 @@ project ('spring-kafka') {
 		}
 		api "org.apache.kafka:kafka-clients:$kafkaVersion"
 		optionalApi "org.apache.kafka:kafka-streams:$kafkaVersion"
+		optionalApi "org.jetbrains.kotlinx:kotlinx-coroutines-reactor:$kotlinCoroutinesVersion"
 		optionalApi 'com.fasterxml.jackson.core:jackson-core'
 		optionalApi 'com.fasterxml.jackson.core:jackson-databind'
 		optionalApi 'com.fasterxml.jackson.datatype:jackson-datatype-jdk8'

spring-kafka-docs/src/main/antora/modules/ROOT/nav.adoc

@@ -11,6 +11,7 @@
 **** xref:kafka/receiving-messages/message-listeners.adoc[]
 **** xref:kafka/receiving-messages/message-listener-container.adoc[]
 **** xref:kafka/receiving-messages/ooo-commits.adoc[]
+**** xref:kafka/receiving-messages/async-returns.adoc[]
 **** xref:kafka/receiving-messages/listener-annotation.adoc[]
 **** xref:kafka/receiving-messages/listener-group-id.adoc[]
 **** xref:kafka/receiving-messages/container-thread-naming.adoc[]
@@ -20,6 +21,7 @@
 **** xref:kafka/receiving-messages/kafkalistener-lifecycle.adoc[]
 **** xref:kafka/receiving-messages/validation.adoc[]
 **** xref:kafka/receiving-messages/rebalance-listeners.adoc[]
+**** xref:kafka/receiving-messages/enforced-rebalance.adoc[]
 **** xref:kafka/receiving-messages/annotation-send-to.adoc[]
 **** xref:kafka/receiving-messages/filtering.adoc[]
 **** xref:kafka/receiving-messages/retrying-deliveries.adoc[]

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/annotation-error-handling.adoc

@@ -451,7 +451,10 @@ AfterRollbackProcessor<String, String> processor =
 When you do not use transactions, you can achieve similar functionality by configuring a `DefaultErrorHandler`.
 See xref:kafka/annotation-error-handling.adoc#error-handlers[Container Error Handlers].
 
-IMPORTANT: Recovery is not possible with a batch listener, since the framework has no knowledge about which record in the batch keeps failing.
+Starting with version 3.2, Recovery can now recover (skip) entire batch of records that keeps failing.
+Set `ContainerProperties.setBatchRecoverAfterRollback(true)` to enable this feature.
+
+IMPORTANT: Default behavior, recovery is not possible with a batch listener, since the framework has no knowledge about which record in the batch keeps failing.
 In such cases, the application listener must handle a record that keeps failing.
 
 See also xref:kafka/annotation-error-handling.adoc#dead-letters[Publishing Dead-letter Records].

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/async-returns.adoc

@@ -0,0 +1,32 @@
+[[async-returns]]
+= Asynchronous `@KafkaListener` Return Types
+
+Starting with version 3.2, `@KafkaListener` (and `@KafkaHandler`) methods can be specified with asynchronous return types, letting the reply be sent asynchronously.
+return types include `CompletableFuture<?>`, `Mono<?>` and Kotlin `suspend` functions.
+
+[source, java]
+----
+@KafkaListener(id = "myListener", topics = "myTopic")
+public CompletableFuture<String> listen(String data) {
+    ...
+    CompletableFuture<String> future = new CompletableFuture<>();
+    future.complete("done");
+    return future;
+}
+----
+
+[source, java]
+----
+@KafkaListener(id = "myListener", topics = "myTopic")
+public Mono<Void> listen(String data) {
+    ...
+    return Mono.empty();
+}
+----
+
+IMPORTANT: The `AckMode` will be automatically set the `MANUAL` and enable out-of-order commits when async return types are detected; instead, the asynchronous completion will ack when the async operation completes.
+When the async result is completed with an error, whether the message is recover or not depends on the container error handler.
+If some exception occurs within the listener method that prevents creation of the async result object, you MUST catch that exception and return an appropriate return object that will cause the message to be ack or recover.
+
+If a `KafkaListenerErrorHandler` is configured on a listener with an async return type (including Kotlin suspend functions), the error handler is invoked after a failure.
+See xref:kafka/annotation-error-handling.adoc[Handling Exceptions] for more information about this error handler and its purpose.

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/enforced-rebalance.adoc

@@ -0,0 +1,34 @@
+[[enforced-rebalance]]
+= Enforcing Consumer Rebalance
+
+Kafka clients now support an option to trigger an https://cwiki.apache.org/confluence/display/KAFKA/KIP-568%3A+Explicit+rebalance+triggering+on+the+Consumer[enforced rebalance].
+Starting with version `3.1.2`, Spring for Apache Kafka provides an option to invoke this API on the Kafka consumer via the message listener container.
+When calling this API, it is simply alerting the Kafka consumer to trigger an enforced rebalance; the actual rebalance will only occur as part of the next `poll()` operation.
+If there is already a rebalance in progress, calling an enforced rebalance is a NO-OP.
+The caller must wait for the current rebalance to complete before invoking another one.
+See the javadocs for `enfroceRebalance` for more details.
+
+The following code snippet shows the essence of enforcing a rebalance using the message listener container.
+
+[source, java]
+----
+@KafkaListener(id = "my.id", topics = "my-topic")
+void listen(ConsumerRecord<String, String> in) {
+    System.out.println("From KafkaListener: " + in);
+}
+
+@Bean
+public ApplicationRunner runner(KafkaTemplate<String, Object> template, KafkaListenerEndpointRegistry registry) {
+    return args -> {
+        final MessageListenerContainer listenerContainer = registry.getListenerContainer("my.id");
+        System.out.println("Enforcing a rebalance");
+        Thread.sleep(5_000);
+        listenerContainer.enforceRebalance();
+        Thread.sleep(5_000);
+    };
+}
+----
+
+As the code above shows, the application uses the `KafkaListenerEndpointRegistry` to gain access to the message listener container and then calling the `enforceRebalnce` API on it.  
+When calling the `enforceRebalance` on the listener container, it delegates the call to the underlying Kafka consumer.
+The Kafka consumer will trigger a rebalance as part of the next `poll()` operation.

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/serdes.adoc

@@ -397,6 +397,8 @@ ConsumerFactory cf = new DefaultKafkaConsumerFactory(myConsumerConfigs,
     new RetryingDeserializer(myUnreliableValueDeserializer, retryTemplate));
 ----
 
+Starting with version `3.1.2`, a `RecoveryCallback` can be set on the `RetryingDeserializer` optionally.
+
 Refer to the https://github.com/spring-projects/spring-retry[spring-retry] project for configuration of the `RetryTemplate` with a retry policy, back off policy, etc.
 
 

spring-kafka-docs/src/main/antora/modules/ROOT/pages/retrytopic/features.adoc

@@ -261,3 +261,41 @@ protected Consumer<DeadLetterPublishingRecovererFactory>
 
 It is recommended that you use the provided resolvers when constructing the custom instance.
 
+[[exc-based-custom-dlt-routing]]
+== Routing of messages to custom DLTs based on thrown exceptions
+
+Starting with version 3.2.0, it's possible to route messages to custom DLTs based on the type of the exception, which has been thrown during their processing.
+In order to do that, there's a need to specify the routing.
+Routing customization consists of the specification of the additional destinations.
+Destinations in turn consist of two settings: the `suffix` and `exceptions`.
+When the exception type specified in `exceptions` has been thrown, the DLT containing the `suffix` will be considered as the target topic for the message before the general purpose DLT is considered.
+Examples of configuration using either annotations or `RetryTopicConfiguration` beans:
+
+[source, java]
+----
+@RetryableTopic(exceptionBasedDltRouting = {
+    @ExceptionBasedDltDestination(
+        suffix = "-deserialization", exceptions = {DeserializationException.class}
+    )}
+)
+@KafkaListener(topics = "my-annotated-topic")
+public void processMessage(MyPojo message) {
+    // ... message processing
+}
+----
+
+[source, java]
+----
+@Bean
+public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
+    return RetryTopicConfigurationBuilder
+            .newInstance()
+            .dltRoutingRules(Map.of("-deserialization", Set.of(DeserializationException.class)))
+            .create(kafkaOperations)
+            .create(template);
+}
+----
+
+`suffix` takes place before the general `dltTopicSuffix` in the custom DLT name.
+Considering presented examples, the message, which caused the `DeserializationException` will be routed to the `my-annotated-topic-deserialization-dlt` instead of the `my-annotated-topic-dlt`.
+Custom DLTs will be created following the same rules as stated in the xref:retrytopic/features.adoc#topics-autocreation[Topics AutoCreation].