Minor fixes and improvements

CONTRIBUTING.adoc

@@ -151,7 +151,7 @@ Please carefully follow the whitespace and formatting conventions already presen
 8. Latin-1 (ISO-8859-1) encoding for Java sources; use `native2ascii` to convert
    if necessary
 
-## Add Apache license header to all new classes
+== Add Apache license header to all new classes
 
 [source, java]
 ----

build.gradle

@@ -384,7 +384,7 @@ task distZip(type: Zip, dependsOn: [docsZip]) { //, schemaZip]) {
 	description = "Builds -${classifier} archive, containing all jars and docs, " +
 			"suitable for community download page."
 
-	ext.baseDir = "${project.name}-${project.version}";
+	ext.baseDir = "${project.name}-${project.version}"
 
 	from('src/dist') {
 		include 'readme.txt'

spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java

@@ -383,8 +383,8 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean {
 
 	public BrokerAddress[] getBrokerAddresses() {
 		List<BrokerAddress> addresses = new ArrayList<BrokerAddress>();
-		for (int i = 0; i < this.kafkaPorts.length; i++) {
-			addresses.add(new BrokerAddress("127.0.0.1", this.kafkaPorts[i]));
+		for (int kafkaPort : this.kafkaPorts) {
+			addresses.add(new BrokerAddress("127.0.0.1", kafkaPort));
 		}
 		return addresses.toArray(new BrokerAddress[0]);
 	}

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

@@ -16,9 +16,9 @@
 
 package org.springframework.kafka.listener;
 
-import java.util.Iterator;
 
 import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 
 import org.springframework.core.log.LogAccessor;
@@ -28,7 +28,6 @@ import org.springframework.core.log.LogAccessor;
  *
  * @author Gary Russell
  * @since 1.1
- *
  */
 public class BatchLoggingErrorHandler implements BatchErrorHandler {
 
@@ -42,9 +41,8 @@ public class BatchLoggingErrorHandler implements BatchErrorHandler {
 			message.append("null ");
 		}
 		else {
-			Iterator<?> iterator = data.iterator();
-			while (iterator.hasNext()) {
-				message.append(iterator.next()).append('\n');
+			for (ConsumerRecord<?, ?> record : data) {
+				message.append(record).append('\n');
 			}
 		}
 		LOGGER.error(thrownException, () -> message.substring(0, message.length() - 1));

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

@@ -231,7 +231,7 @@ public class ContainerProperties {
 	 */
 	public ContainerProperties(String... topics) {
 		Assert.notEmpty(topics, "An array of topics must be provided");
-		this.topics = Arrays.asList(topics).toArray(new String[topics.length]);
+		this.topics = topics.clone();
 		this.topicPattern = null;
 		this.topicPartitions = null;
 	}

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

@@ -126,12 +126,10 @@ public class DeadLetterPublishingRecoverer implements BiConsumer<ConsumerRecord<
 		this.templates = templates;
 		this.transactional = templates.values().iterator().next().isTransactional();
 		Boolean tx = this.transactional;
-		Assert.isTrue(!templates.values()
+		Assert.isTrue(templates.values()
 			.stream()
 			.map(t -> t.isTransactional())
-			.filter(t -> !t.equals(tx))
-			.findFirst()
-			.isPresent(), "All templates must have the same setting for transactional");
+			.allMatch(t -> t.equals(tx)), "All templates must have the same setting for transactional");
 		this.destinationResolver = destinationResolver;
 	}
 
@@ -172,7 +170,7 @@ public class DeadLetterPublishingRecoverer implements BiConsumer<ConsumerRecord<
 		if (key.isPresent()) {
 			return (KafkaTemplate<Object, Object>) this.templates.get(key.get());
 		}
-		LOGGER.warn(() -> "Failed to find a template for " + value.getClass() + " attemting to use the last entry");
+		LOGGER.warn(() -> "Failed to find a template for " + value.getClass() + " attempting to use the last entry");
 		return (KafkaTemplate<Object, Object>) this.templates.values()
 				.stream()
 				.reduce((first,  second) -> second)

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

@@ -1344,7 +1344,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 				while (iterator.hasNext()) {
 					records.add(iterator.next());
 				}
-				((RemainingRecordsErrorHandler) this.errorHandler).handle(decorateException(e), records, this.consumer,
+				this.errorHandler.handle(decorateException(e), records, this.consumer,
 						KafkaMessageListenerContainer.this.container);
 			}
 			else {

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java

@@ -288,7 +288,7 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
 			if (this.hasAckParameter && acknowledgment == null) {
 				throw new ListenerExecutionFailedException("invokeHandler Failed",
 						new IllegalStateException("No Acknowledgment available as an argument, "
-								+ "the listener container must have a MANUAL Ackmode to populate the Acknowledgment.",
+								+ "the listener container must have a MANUAL AckMode to populate the Acknowledgment.",
 								ex));
 			}
 			throw new ListenerExecutionFailedException(createMessagingErrorMessage("Listener method could not " +

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

@@ -20,7 +20,7 @@ import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 
 /**
- * Result for a Listenablefuture after a send.
+ * Result for a ListenableFuture after a send.
  *
  * @param <K> the key type.
  * @param <V> the value type.

spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessageConverter.java

@@ -36,9 +36,8 @@ import org.springframework.lang.Nullable;
 public interface MessageConverter {
 
 	@Nullable
-	static String getGroupid() {
-		String groupId = KafkaUtils.getConsumerGroupId();
-		return groupId == null ? null : groupId;
+	static String getGroupId() {
+		return KafkaUtils.getConsumerGroupId();
 	}
 
 	default void commonHeaders(Acknowledgment acknowledgment, Consumer<?, ?> consumer, Map<String, Object> rawHeaders,
@@ -52,7 +51,7 @@ public interface MessageConverter {
 		rawHeaders.put(KafkaHeaders.TIMESTAMP_TYPE, timestampType);
 		rawHeaders.put(KafkaHeaders.RECEIVED_TIMESTAMP, timestamp);
 		JavaUtils.INSTANCE
-			.acceptIfNotNull(KafkaHeaders.GROUP_ID, MessageConverter.getGroupid(),
+			.acceptIfNotNull(KafkaHeaders.GROUP_ID, MessageConverter.getGroupId(),
 					(key, val) -> rawHeaders.put(key, val))
 			.acceptIfNotNull(KafkaHeaders.ACKNOWLEDGMENT, acknowledgment, (key, val) -> rawHeaders.put(key, val))
 			.acceptIfNotNull(KafkaHeaders.CONSUMER, consumer, (key, val) -> rawHeaders.put(key, val));

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -680,7 +680,7 @@ public class EnableKafkaIntegrationTests {
 		assertThat(this.config.badAckException).isInstanceOf(IllegalStateException.class);
 		assertThat(this.config.badAckException.getMessage())
 				.isEqualTo("No Acknowledgment available as an argument, "
-						+ "the listener container must have a MANUAL Ackmode to populate the Acknowledgment.");
+						+ "the listener container must have a MANUAL AckMode to populate the Acknowledgment.");
 	}
 
 	@Test

src/reference/asciidoc/kafka.adoc

@@ -791,7 +791,7 @@ This prevents the container from starting if any of the configured topics are no
 It does not apply if the container is configured to listen to a topic pattern (regex).
 Previously, the container threads looped within the `consumer.poll()` method waiting for the topic to appear while logging many messages.
 Aside from the logs, there was no indication that there was a problem.
-To restore the previous behavior, you canset the property to `false`.
+To restore the previous behavior, you can set the property to `false`.
 
 [[using-ConcurrentMessageListenerContainer]]
 ====== Using `ConcurrentMessageListenerContainer`
@@ -1455,7 +1455,7 @@ public class Config implements KafkaListenerConfigurer {
 ----
 ====
 
-The follwing examples show how to validate:
+The following examples show how to validate:
 
 ====
 [source, java]

src/reference/asciidoc/si-kafka.adoc

@@ -375,7 +375,7 @@ Refer to the javadocs for available properties.
 
 [[max-poll-records]]
 By default, `max.poll.records` must be either explicitly set in the consumer factory, or it will be forced to 1 if the consumer factory is a `DefaultKafkaConsumerFactory`.
-Starting with versio 3.2, you can set the property `allowMultiFetch` to `true` to override this behavior.
+Starting with version 3.2, you can set the property `allowMultiFetch` to `true` to override this behavior.
 
 IMPORTANT: You must poll the consumer within `max.poll.interval.ms` to avoid a rebalance.
 If you set `allowMultiFetch` to `true` you must process all the retrieved records, and poll again, within `max.poll.interval.ms`.