Minor fixes and improvements

Author: neetkee

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 BrokerAddress getBrokerAddress(int i) {
 
 	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 @@
  *
  * @author Gary Russell
  * @since 1.1
- *
  */
 public class BatchLoggingErrorHandler implements BatchErrorHandler {
 
@@ -42,9 +41,8 @@ public void handle(Exception thrownException, ConsumerRecords<?, ?> data) {
 			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 enum AckMode {
 	 */
 	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 DeadLetterPublishingRecoverer(Map<Class<?>, KafkaTemplate<? extends Objec
 		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 @@ private KafkaTemplate<Object, Object> findTemplateForValue(Object value) {
 		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 @@ private void invokeErrorHandler(final ConsumerRecord<K, V> record,
 				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 @@ protected final Object invokeHandler(Object data, Acknowledgment acknowledgment,
 			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.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 @@
 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 @@ default void commonHeaders(Acknowledgment acknowledgment, Consumer<?, ?> consume
 		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));