From 2c5914ba662022e10cff809c8f70c72514f37201 Mon Sep 17 00:00:00 2001 From: Artem Yakshin Date: Sat, 27 Apr 2019 15:54:42 +0300 Subject: [PATCH] Minor fixes and improvements --- CONTRIBUTING.adoc | 2 +- build.gradle | 2 +- .../springframework/kafka/test/EmbeddedKafkaBroker.java | 4 ++-- .../kafka/listener/BatchLoggingErrorHandler.java | 8 +++----- .../kafka/listener/ContainerProperties.java | 2 +- .../kafka/listener/DeadLetterPublishingRecoverer.java | 8 +++----- .../kafka/listener/KafkaMessageListenerContainer.java | 2 +- .../listener/adapter/MessagingMessageListenerAdapter.java | 2 +- .../org/springframework/kafka/support/SendResult.java | 2 +- .../kafka/support/converter/MessageConverter.java | 7 +++---- .../kafka/annotation/EnableKafkaIntegrationTests.java | 2 +- src/reference/asciidoc/kafka.adoc | 4 ++-- src/reference/asciidoc/si-kafka.adoc | 2 +- 13 files changed, 21 insertions(+), 26 deletions(-) diff --git a/CONTRIBUTING.adoc b/CONTRIBUTING.adoc index 2c311b45..44aaf091 100644 --- a/CONTRIBUTING.adoc +++ b/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] ---- diff --git a/build.gradle b/build.gradle index 62c76b50..c727fbaf 100644 --- a/build.gradle +++ b/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' diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java index 591dcc8f..1a660c29 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java +++ b/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 addresses = new ArrayList(); - 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]); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/BatchLoggingErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/BatchLoggingErrorHandler.java index 9b477c59..55c7895d 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/BatchLoggingErrorHandler.java +++ b/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)); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java index 3c44827a..2373cc31 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java +++ b/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; } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java index e52ecac5..376927bc 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java @@ -126,12 +126,10 @@ public class DeadLetterPublishingRecoverer implements BiConsumer 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) 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) this.templates.values() .stream() .reduce((first, second) -> second) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index 44478392..ff8d8e30 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -1344,7 +1344,7 @@ public class KafkaMessageListenerContainer // 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 { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java index 14664415..8264dc57 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java @@ -288,7 +288,7 @@ public abstract class MessagingMessageListenerAdapter 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 " + diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/SendResult.java b/spring-kafka/src/main/java/org/springframework/kafka/support/SendResult.java index be441c44..c85f5c12 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/SendResult.java +++ b/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 the key type. * @param the value type. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessageConverter.java b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessageConverter.java index cace92a3..44262ee0 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessageConverter.java +++ b/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 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)); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java index 12101aac..c76f1f00 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java +++ b/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 diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index f7ab2152..f8271008 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/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] diff --git a/src/reference/asciidoc/si-kafka.adoc b/src/reference/asciidoc/si-kafka.adoc index c70c1ed5..6ed88184 100644 --- a/src/reference/asciidoc/si-kafka.adoc +++ b/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`.