Merge branch 'main' into spring-projectsGH-2588

Author: Zhiyang.Wang1

build.gradle

@@ -65,17 +65,17 @@ ext {
 	junit4Version = '4.13.2'
 	junitJupiterVersion = '5.10.0'
 	kafkaVersion = '3.6.0'
-	log4jVersion = '2.20.0'
+	log4jVersion = '2.21.0'
 	micrometerDocsVersion = "1.0.2"
-	micrometerVersion = '1.12.0-RC1'
-	micrometerTracingVersion = '1.2.0-RC1'
+	micrometerVersion = '1.12.0'
+	micrometerTracingVersion = '1.2.0'
 	mockitoVersion = '5.6.0'
-	reactorVersion = '2023.0.0-RC1'
+	reactorVersion = '2023.0.0'
 	scalaVersion = '2.13'
-	springBootVersion = '3.0.9' // docs module
-	springDataVersion = '2023.1.0-RC1'
+	springBootVersion = '3.0.12' // docs module
+	springDataVersion = '2023.1.0'
 	springRetryVersion = '2.0.4'
-	springVersion = '6.1.0-SNAPSHOT'
+	springVersion = '6.1.0'
 	zookeeperVersion = '3.8.3'
 
 	idPrefix = 'kafka'

gradle.properties

@@ -1,4 +1,4 @@
-version=3.1.0-SNAPSHOT
+version=3.1.1-SNAPSHOT
 org.gradle.jvmargs=-Xmx1536M -Dfile.encoding=UTF-8
 org.gradle.caching=true
 org.gradle.parallel=true

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/message-listener-container.adoc

@@ -233,6 +233,8 @@ With a record listener, when `nack()` is called, any pending offsets are committ
 The consumer can be paused before redelivery, by setting the `sleep` argument.
 This is similar functionality to throwing an exception when the container is configured with a `DefaultErrorHandler`.
 
+IMPORTANT: `nack()` pauses the entire listener for the specified sleep duration including all assigned partitions.
+
 When using a batch listener, you can specify the index within the batch where the failure occurred.
 When `nack()` is called, offsets will be committed for records before the index and seeks are performed on the partitions for the failed and discarded records so that they will be redelivered on the next `poll()`.
 

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/ooo-commits.adoc

@@ -9,3 +9,5 @@ The consumer will be paused (no new records delivered) until all the offsets for
 
 IMPORTANT: While this feature allows applications to process records asynchronously, it should be understood that it increases the possibility of duplicate deliveries after a failure.
 
+IMPORTANT: When `asyncAcks` is activated, it is not possible to use `nack()` (negative acknowledgments) when xref:kafka/receiving-messages/message-listener-container.adoc#committing-offsets[Committing Offsets].
+

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

@@ -339,8 +339,8 @@ public class EmbeddedKafkaConditionTests {
 }
 ----
 
-A stand-alone (not Spring test context) broker will be created if the class annotated with `@EmbeddedBroker` is not also annotated (or meta annotated) with `ExtendedWith(SpringExtension.class)`.
-`@SpringJunitConfig` and `@SpringBootTest` are so meta annotated and the context-based broker will be used when either of those annotations are also present.
+A standalone broker (outside the Spring's TestContext) will be created unless a class annotated `@EmbeddedKafka` is also annotated (or meta-annotated) with `ExtendWith(SpringExtension.class)`.
+`@SpringJunitConfig` and `@SpringBootTest` are so meta-annotated and the context-based broker will be used when either of those annotations are also present.
 
 IMPORTANT: When there is a Spring test application context available, the topics and broker properties can contain property placeholders, which will be resolved as long as the property is defined somewhere.
 If there is no Spring context available, these placeholders won't be resolved.

spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlingUtils.java

@@ -22,6 +22,8 @@
 import java.util.Properties;
 import java.util.Set;
 import java.util.function.BiConsumer;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -189,12 +191,9 @@ private static void listen(List<RetryListener> listeners, ConsumerRecords<?, ?>
 	 * @return the String.
 	 */
 	public static String recordsToString(ConsumerRecords<?, ?> records) {
-		StringBuffer sb = new StringBuffer();
-		records.spliterator().forEachRemaining(rec -> sb
-				.append(KafkaUtils.format(rec))
-				.append(','));
-		sb.deleteCharAt(sb.length() - 1);
-		return sb.toString();
+		return StreamSupport.stream(records.spliterator(), false)
+				.map(KafkaUtils::format)
+				.collect(Collectors.joining(","));
 	}
 
 	/**

spring-kafka/src/main/java/org/springframework/kafka/support/Acknowledgment.java

@@ -41,7 +41,9 @@ public interface Acknowledgment {
 	/**
 	 * Negatively acknowledge the current record - discard remaining records from the poll
 	 * and re-seek all partitions so that this record will be redelivered after the sleep
-	 * duration. Must be called on the consumer thread.
+	 * duration. This will pause reading for the entire message listener for the specified
+	 * sleep duration and is not limited to a single partition.
+	 * Must be called on the consumer thread.
 	 * <p>
 	 * @param sleep the duration to sleep; the actual sleep time will be larger of this value
 	 * and the container's {@code maxPollInterval}, which defaults to 5 seconds.

spring-kafka/src/main/java/org/springframework/kafka/support/LoggingProducerListener.java

@@ -69,7 +69,7 @@ public void setMaxContentLogged(int maxContentLogged) {
 	@Override
 	public void onError(ProducerRecord<K, V> record, @Nullable RecordMetadata recordMetadata, Exception exception) {
 		this.logger.error(exception, () -> {
-			StringBuffer logOutput = new StringBuffer();
+			StringBuilder logOutput = new StringBuilder();
 			logOutput.append("Exception thrown when sending a message");
 			if (this.includeContents) {
 				logOutput.append(" with key='")