From 9a73a4a2021b0ceaaffb62cc1759338c78124754 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 15 Jan 2019 16:15:21 -0500 Subject: [PATCH] GH-787: Fix upper-bounds for generics (#937) * GH-787: Fix upper-bounds for generics Fixes spring-projects/spring-kafka#787 * Allow to configure listener container with strategies based on the super classes of key and value generics * * Revert some code style change * Update Copyrights * Use `Supplier` for non-constant asserts --- .../EmbeddedKafkaContextCustomizerTests.java | 23 +++- ...AbstractKafkaListenerContainerFactory.java | 22 ++-- .../config/AbstractKafkaListenerEndpoint.java | 15 ++- .../AbstractMessageListenerContainer.java | 48 ++++--- .../ConcurrentMessageListenerContainer.java | 28 ++--- .../KafkaMessageListenerContainer.java | 117 +++++++++--------- .../EnableKafkaIntegrationTests.java | 12 +- 7 files changed, 136 insertions(+), 129 deletions(-) diff --git a/spring-kafka-test/src/test/java/org/springframework/kafka/test/context/EmbeddedKafkaContextCustomizerTests.java b/spring-kafka-test/src/test/java/org/springframework/kafka/test/context/EmbeddedKafkaContextCustomizerTests.java index c4974e1b..ceaf1edf 100644 --- a/spring-kafka-test/src/test/java/org/springframework/kafka/test/context/EmbeddedKafkaContextCustomizerTests.java +++ b/spring-kafka-test/src/test/java/org/springframework/kafka/test/context/EmbeddedKafkaContextCustomizerTests.java @@ -32,40 +32,46 @@ import org.springframework.core.env.ConfigurableEnvironment; import org.springframework.kafka.test.EmbeddedKafkaBroker; - /** * @author Oleg Artyomov * @author Sergio Lourenco + * @author Artem Bilan + * * @since 1.3 */ public class EmbeddedKafkaContextCustomizerTests { private EmbeddedKafka annotationFromFirstClass; + private EmbeddedKafka annotationFromSecondClass; @Before public void beforeEachTest() { annotationFromFirstClass = AnnotationUtils.findAnnotation(TestWithEmbeddedKafka.class, EmbeddedKafka.class); - annotationFromSecondClass = AnnotationUtils.findAnnotation(SecondTestWithEmbeddedKafka.class, EmbeddedKafka.class); + annotationFromSecondClass = + AnnotationUtils.findAnnotation(SecondTestWithEmbeddedKafka.class, EmbeddedKafka.class); } @Test public void testHashCode() { assertThat(new EmbeddedKafkaContextCustomizer(annotationFromFirstClass).hashCode()).isNotEqualTo(0); - assertThat(new EmbeddedKafkaContextCustomizer(annotationFromFirstClass).hashCode()).isEqualTo(new EmbeddedKafkaContextCustomizer(annotationFromSecondClass).hashCode()); + assertThat(new EmbeddedKafkaContextCustomizer(annotationFromFirstClass).hashCode()) + .isEqualTo(new EmbeddedKafkaContextCustomizer(annotationFromSecondClass).hashCode()); } @Test public void testEquals() { - assertThat(new EmbeddedKafkaContextCustomizer(annotationFromFirstClass)).isEqualTo(new EmbeddedKafkaContextCustomizer(annotationFromSecondClass)); + assertThat(new EmbeddedKafkaContextCustomizer(annotationFromFirstClass)) + .isEqualTo(new EmbeddedKafkaContextCustomizer(annotationFromSecondClass)); assertThat(new EmbeddedKafkaContextCustomizer(annotationFromFirstClass)).isNotEqualTo(new Object()); } @Test public void testPorts() { - EmbeddedKafka annotationWithPorts = AnnotationUtils.findAnnotation(TestWithEmbeddedKafkaPorts.class, EmbeddedKafka.class); + EmbeddedKafka annotationWithPorts = + AnnotationUtils.findAnnotation(TestWithEmbeddedKafkaPorts.class, EmbeddedKafka.class); EmbeddedKafkaContextCustomizer customizer = new EmbeddedKafkaContextCustomizer(annotationWithPorts); ConfigurableApplicationContext context = mock(ConfigurableApplicationContext.class); BeanFactoryStub factoryStub = new BeanFactoryStub(); @@ -73,7 +79,8 @@ public class EmbeddedKafkaContextCustomizerTests { given(context.getEnvironment()).willReturn(mock(ConfigurableEnvironment.class)); customizer.customizeContext(context, null); - assertThat(factoryStub.getBroker().getBrokersAsString()).isEqualTo("127.0.0.1:" + annotationWithPorts.ports()[0]); + assertThat(factoryStub.getBroker().getBrokersAsString()) + .isEqualTo("127.0.0.1:" + annotationWithPorts.ports()[0]); } @@ -92,7 +99,9 @@ public class EmbeddedKafkaContextCustomizerTests { } + @SuppressWarnings("serial") private class BeanFactoryStub extends DefaultListableBeanFactory { + private Object bean; public EmbeddedKafkaBroker getBroker() { @@ -114,5 +123,7 @@ public class EmbeddedKafkaContextCustomizerTests { public void registerDisposableBean(String beanName, DisposableBean bean) { } + } + } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java index 9fbc8de7..c7e46687 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2018 the original author or authors. + * Copyright 2014-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -61,7 +61,7 @@ public abstract class AbstractKafkaListenerContainerFactory errorHandler; - private ConsumerFactory consumerFactory; + private ConsumerFactory consumerFactory; private Boolean autoStartup; @@ -69,7 +69,7 @@ public abstract class AbstractKafkaListenerContainerFactory recordFilterStrategy; + private RecordFilterStrategy recordFilterStrategy; private Boolean ackDiscarded; @@ -85,7 +85,7 @@ public abstract class AbstractKafkaListenerContainerFactory replyTemplate; - private AfterRollbackProcessor afterRollbackProcessor; + private AfterRollbackProcessor afterRollbackProcessor; private ReplyHeadersConfigurer replyHeadersConfigurer; @@ -93,11 +93,11 @@ public abstract class AbstractKafkaListenerContainerFactory consumerFactory) { + public void setConsumerFactory(ConsumerFactory consumerFactory) { this.consumerFactory = consumerFactory; } - public ConsumerFactory getConsumerFactory() { + public ConsumerFactory getConsumerFactory() { return this.consumerFactory; } @@ -131,7 +131,7 @@ public abstract class AbstractKafkaListenerContainerFactory recordFilterStrategy) { + public void setRecordFilterStrategy(RecordFilterStrategy recordFilterStrategy) { this.recordFilterStrategy = recordFilterStrategy; } @@ -231,7 +231,7 @@ public abstract class AbstractKafkaListenerContainerFactory afterRollbackProcessor) { + public void setAfterRollbackProcessor(AfterRollbackProcessor afterRollbackProcessor) { this.afterRollbackProcessor = afterRollbackProcessor; } @@ -258,11 +258,13 @@ public abstract class AbstractKafkaListenerContainerFactory "The error handler must be a BatchErrorHandler, not " + + this.errorHandler.getClass().getName()); } else { Assert.state(this.errorHandler instanceof ErrorHandler, - "The error handler must be an ErrorHandler, not " + this.errorHandler.getClass().getName()); + () -> "The error handler must be an ErrorHandler, not " + + this.errorHandler.getClass().getName()); } } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java index 58da5dd3..1611c546 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2018 the original author or authors. + * Copyright 2014-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -268,7 +268,7 @@ public abstract class AbstractKafkaListenerEndpoint return this.replyTemplate; } - protected RecordFilterStrategy getRecordFilterStrategy() { + protected RecordFilterStrategy getRecordFilterStrategy() { return this.recordFilterStrategy; } @@ -276,8 +276,9 @@ public abstract class AbstractKafkaListenerEndpoint * Set a {@link RecordFilterStrategy} implementation. * @param recordFilterStrategy the strategy implementation. */ - public void setRecordFilterStrategy(RecordFilterStrategy recordFilterStrategy) { - this.recordFilterStrategy = recordFilterStrategy; + @SuppressWarnings("unchecked") + public void setRecordFilterStrategy(RecordFilterStrategy recordFilterStrategy) { + this.recordFilterStrategy = (RecordFilterStrategy) recordFilterStrategy; } protected boolean isAckDiscarded() { @@ -285,8 +286,7 @@ public abstract class AbstractKafkaListenerEndpoint } /** - * Set to true if the {@link #setRecordFilterStrategy(RecordFilterStrategy) - * recordFilterStrategy} is in use. + * Set to true if the {@link #setRecordFilterStrategy(RecordFilterStrategy)} is in use. * @param ackDiscarded the ackDiscarded. */ public void setAckDiscarded(boolean ackDiscarded) { @@ -310,8 +310,7 @@ public abstract class AbstractKafkaListenerEndpoint } /** - * Set a callback to be used with the {@link #setRetryTemplate(RetryTemplate) - * retryTemplate}. + * Set a callback to be used with the {@link #setRetryTemplate(RetryTemplate)}. * @param recoveryCallback the callback. */ public void setRecoveryCallback(RecoveryCallback recoveryCallback) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java index 8a302963..c78b916d 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2018 the original author or authors. + * Copyright 2016-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -36,6 +36,7 @@ import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.event.ContainerStoppedEvent; +import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -76,7 +77,8 @@ public abstract class AbstractMessageListenerContainer private int phase = DEFAULT_PHASE; - private AfterRollbackProcessor afterRollbackProcessor = new DefaultAfterRollbackProcessor<>(); + private AfterRollbackProcessor afterRollbackProcessor = + new DefaultAfterRollbackProcessor<>(); private volatile boolean running = false; @@ -98,11 +100,12 @@ public abstract class AbstractMessageListenerContainer * @param consumerFactory the factory. * @param containerProperties the properties. */ - protected AbstractMessageListenerContainer(ConsumerFactory consumerFactory, + @SuppressWarnings("unchecked") + protected AbstractMessageListenerContainer(ConsumerFactory consumerFactory, ContainerProperties containerProperties) { Assert.notNull(containerProperties, "'containerProperties' cannot be null"); - this.consumerFactory = consumerFactory; + this.consumerFactory = (ConsumerFactory) consumerFactory; if (containerProperties.getTopics() != null) { this.containerProperties = new ContainerProperties(containerProperties.getTopics()); } @@ -221,7 +224,7 @@ public abstract class AbstractMessageListenerContainer return this.phase; } - protected AfterRollbackProcessor getAfterRollbackProcessor() { + protected AfterRollbackProcessor getAfterRollbackProcessor() { return this.afterRollbackProcessor; } @@ -232,7 +235,7 @@ public abstract class AbstractMessageListenerContainer * @param afterRollbackProcessor the processor. * @since 1.3.5 */ - public void setAfterRollbackProcessor(AfterRollbackProcessor afterRollbackProcessor) { + public void setAfterRollbackProcessor(AfterRollbackProcessor afterRollbackProcessor) { Assert.notNull(afterRollbackProcessor, "'afterRollbackProcessor' cannot be null"); this.afterRollbackProcessor = afterRollbackProcessor; } @@ -252,9 +255,8 @@ public abstract class AbstractMessageListenerContainer checkGroupId(); synchronized (this.lifecycleMonitor) { if (!isRunning()) { - Assert.isTrue( - this.containerProperties.getMessageListener() instanceof GenericMessageListener, - "A " + GenericMessageListener.class.getName() + " implementation must be provided"); + Assert.isTrue(this.containerProperties.getMessageListener() instanceof GenericMessageListener, + () -> "A " + GenericMessageListener.class.getName() + " implementation must be provided"); doStart(); } } @@ -262,13 +264,14 @@ public abstract class AbstractMessageListenerContainer protected void checkTopics() { if (this.containerProperties.isMissingTopicsFatal() && this.containerProperties.getTopicPattern() == null) { - try (Consumer consumer = this.consumerFactory.createConsumer(this.containerProperties.getGroupId(), - this.containerProperties.getClientId(), null)) { + try (Consumer consumer = + this.consumerFactory.createConsumer(this.containerProperties.getGroupId(), + this.containerProperties.getClientId(), null)) { if (consumer != null) { String[] topics = this.containerProperties.getTopics(); if (topics == null) { topics = Arrays.stream(this.containerProperties.getTopicPartitions()) - .map(tp -> tp.topic()) + .map(TopicPartitionInitialOffset::topic) .toArray(String[]::new); } List missing = new ArrayList<>(); @@ -293,12 +296,12 @@ public abstract class AbstractMessageListenerContainer if (this.consumerFactory != null) { // we always have one for standard containers Object groupIdConfig = this.consumerFactory.getConfigurationProperties() .get(ConsumerConfig.GROUP_ID_CONFIG); - hasGroupIdConsumerConfig = groupIdConfig != null && groupIdConfig instanceof String - && StringUtils.hasText((String) groupIdConfig); + hasGroupIdConsumerConfig = + groupIdConfig instanceof String && StringUtils.hasText((String) groupIdConfig); } Assert.state(hasGroupIdConsumerConfig || StringUtils.hasText(this.containerProperties.getGroupId()), "No group.id found in consumer config, container properties, or @KafkaListener annotation; " - + "a group.id is required when group management is used."); + + "a group.id is required when group management is used."); } } @@ -309,13 +312,7 @@ public abstract class AbstractMessageListenerContainer synchronized (this.lifecycleMonitor) { if (isRunning()) { final CountDownLatch latch = new CountDownLatch(1); - doStop(new Runnable() { - - @Override - public void run() { - latch.countDown(); - } - }); + doStop(latch::countDown); try { latch.await(this.containerProperties.getShutdownTimeout(), TimeUnit.MILLISECONDS); // NOSONAR publishContainerStoppedEvent(); @@ -370,13 +367,14 @@ public abstract class AbstractMessageListenerContainer } protected void publishContainerStoppedEvent() { - if (getApplicationEventPublisher() != null) { - getApplicationEventPublisher().publishEvent(new ContainerStoppedEvent(this, parentOrThis())); + ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher(); + if (applicationEventPublisher != null) { + applicationEventPublisher.publishEvent(new ContainerStoppedEvent(this, parentOrThis())); } } /** - * Ruturn this or a parent container if this has a parent. + * Return this or a parent container if this has a parent. * @return the parent or this. * @since 2.2.1 */ diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java index 1f4687cd..9b655bc3 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2018 the original author or authors. + * Copyright 2015-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -65,8 +65,9 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis * @param consumerFactory the consumer factory. * @param containerProperties the container properties. */ - public ConcurrentMessageListenerContainer(ConsumerFactory consumerFactory, + public ConcurrentMessageListenerContainer(ConsumerFactory consumerFactory, ContainerProperties containerProperties) { + super(consumerFactory, containerProperties); Assert.notNull(consumerFactory, "A ConsumerFactory must be provided"); } @@ -100,7 +101,7 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis return this.containers.stream() .map(KafkaMessageListenerContainer::getAssignedPartitions) .filter(Objects::nonNull) - .flatMap(assignedPartitions -> assignedPartitions.stream()) + .flatMap(Collection::stream) .collect(Collectors.toList()); } @@ -135,8 +136,7 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis checkTopics(); ContainerProperties containerProperties = getContainerProperties(); TopicPartitionInitialOffset[] topicPartitions = containerProperties.getTopicPartitions(); - if (topicPartitions != null - && this.concurrency > topicPartitions.length) { + if (topicPartitions != null && this.concurrency > topicPartitions.length) { this.logger.warn("When specific partitions are provided, the concurrency must be less than or " + "equal to the number of partitions; reduced from " + this.concurrency + " to " + topicPartitions.length); @@ -147,8 +147,7 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis for (int i = 0; i < this.concurrency; i++) { KafkaMessageListenerContainer container; if (topicPartitions == null) { - container = new KafkaMessageListenerContainer<>(this, this.consumerFactory, - containerProperties); + container = new KafkaMessageListenerContainer<>(this, this.consumerFactory, containerProperties); } else { container = new KafkaMessageListenerContainer<>(this, this.consumerFactory, @@ -213,15 +212,10 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis } for (KafkaMessageListenerContainer container : this.containers) { if (container.isRunning()) { - container.stop(new Runnable() { - - @Override - public void run() { - if (count.decrementAndGet() <= 0) { - callback.run(); - } + container.stop(() -> { + if (count.decrementAndGet() <= 0) { + callback.run(); } - }); } } @@ -232,13 +226,13 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis @Override public void pause() { super.pause(); - this.containers.forEach(c -> c.pause()); + this.containers.forEach(AbstractMessageListenerContainer::pause); } @Override public void resume() { super.resume(); - this.containers.forEach(c -> c.resume()); + this.containers.forEach(AbstractMessageListenerContainer::resume); } @Override 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 ae57d502..3201e287 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 @@ -27,7 +27,6 @@ import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.Iterator; -import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Map.Entry; @@ -37,6 +36,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ScheduledFuture; import java.util.stream.Collectors; +import java.util.stream.StreamSupport; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -112,7 +112,8 @@ import org.springframework.util.concurrent.ListenableFutureCallback; * @author Yang Qiju * @author Tom van den Berge */ -public class KafkaMessageListenerContainer extends AbstractMessageListenerContainer { // NOSONAR comment density +public class KafkaMessageListenerContainer + extends AbstractMessageListenerContainer { // NOSONAR comment density private static final int DEFAULT_ACK_TIME = 5000; @@ -135,8 +136,9 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener * @param consumerFactory the consumer factory. * @param containerProperties the container properties. */ - public KafkaMessageListenerContainer(ConsumerFactory consumerFactory, + public KafkaMessageListenerContainer(ConsumerFactory consumerFactory, ContainerProperties containerProperties) { + this(null, consumerFactory, containerProperties, (TopicPartitionInitialOffset[]) null); } @@ -147,8 +149,9 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener * @param containerProperties the container properties. * @param topicPartitions the topics/partitions; duplicates are eliminated. */ - public KafkaMessageListenerContainer(ConsumerFactory consumerFactory, + public KafkaMessageListenerContainer(ConsumerFactory consumerFactory, ContainerProperties containerProperties, TopicPartitionInitialOffset... topicPartitions) { + this(null, consumerFactory, containerProperties, topicPartitions); } @@ -159,8 +162,9 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener * @param containerProperties the container properties. */ KafkaMessageListenerContainer(AbstractMessageListenerContainer container, - ConsumerFactory consumerFactory, + ConsumerFactory consumerFactory, ContainerProperties containerProperties) { + this(container, consumerFactory, containerProperties, (TopicPartitionInitialOffset[]) null); } @@ -173,8 +177,9 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener * @param topicPartitions the topics/partitions; duplicates are eliminated. */ KafkaMessageListenerContainer(AbstractMessageListenerContainer container, - ConsumerFactory consumerFactory, + ConsumerFactory consumerFactory, ContainerProperties containerProperties, TopicPartitionInitialOffset... topicPartitions) { + super(consumerFactory, containerProperties); Assert.notNull(consumerFactory, "A ConsumerFactory must be provided"); this.container = container == null ? this : container; @@ -354,6 +359,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener this.logger.error("Failed to publish consumer stopping event", e); } } + private void publishConsumerStoppedEvent() { if (getApplicationEventPublisher() != null) { getApplicationEventPublisher().publishEvent(new ConsumerStoppedEvent(this, this.container)); @@ -483,7 +489,8 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener @SuppressWarnings("unchecked") ListenerConsumer(GenericMessageListener listener, ListenerType listenerType) { Assert.state(!this.isAnyManualAck || !this.autoCommit, - "Consumer cannot be configured for auto commit for ackMode " + this.containerProperties.getAckMode()); + () -> "Consumer cannot be configured for auto commit for ackMode " + + this.containerProperties.getAckMode()); this.consumer = KafkaMessageListenerContainer.this.consumerFactory.createConsumer( this.consumerGroupId, @@ -530,7 +537,8 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener this.errorHandler = determineErrorHandler(errHandler); this.batchErrorHandler = new BatchLoggingErrorHandler(); } - Assert.state(!this.isBatchListener || !this.isRecordAck, "Cannot use AckMode.RECORD with a batch listener"); + Assert.state(!this.isBatchListener || !this.isRecordAck, + "Cannot use AckMode.RECORD with a batch listener"); if (this.containerProperties.getScheduler() != null) { this.taskScheduler = this.containerProperties.getScheduler(); this.taskSchedulerExplicitlySet = true; @@ -540,12 +548,13 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener threadPoolTaskScheduler.initialize(); this.taskScheduler = threadPoolTaskScheduler; } - this.monitorTask = this.taskScheduler.scheduleAtFixedRate(() -> checkConsumer(), + this.monitorTask = this.taskScheduler.scheduleAtFixedRate(this::checkConsumer, this.containerProperties.getMonitorInterval() * 1000); // NOSONAR magic # if (this.containerProperties.isLogContainerConfig()) { this.logger.info(this); } - Map props = KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties(); + Map props = + KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties(); this.checkNullKeyForExceptions = checkDeserializer(findDeserializerClass(props, false)); this.checkNullValueForExceptions = checkDeserializer(findDeserializerClass(props, true)); } @@ -564,7 +573,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } } - private void subscribeOrAssignTopics(final Consumer consumer) { + private void subscribeOrAssignTopics(final Consumer consumer) { if (KafkaMessageListenerContainer.this.topicPartitions == null) { ConsumerRebalanceListener rebalanceListener = new ListenerConsumerRebalanceListener(); if (this.containerProperties.getTopicPattern() != null) { @@ -590,9 +599,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener private boolean checkDeserializer(Object deser) { return deser instanceof Class ? ErrorHandlingDeserializer2.class.isAssignableFrom((Class) deser) - : deser instanceof String - ? ((String) deser).equals(ErrorHandlingDeserializer2.class.getName()) - : false; + : deser instanceof String && deser.equals(ErrorHandlingDeserializer2.class.getName()); } protected void checkConsumer() { @@ -940,7 +947,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } } - @SuppressWarnings({ RAW_TYPES }) + @SuppressWarnings({ "unchecked", RAW_TYPES }) private void invokeBatchListenerInTx(final ConsumerRecords records, final List> recordList) { try { @@ -952,7 +959,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener if (ListenerConsumer.this.kafkaTxManager != null) { producer = ((KafkaResourceHolder) TransactionSynchronizationManager .getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory())) - .getProducer(); // NOSONAR nullable + .getProducer(); // NOSONAR nullable } RuntimeException aborted = doInvokeBatchListener(records, recordList, producer); if (aborted != null) { @@ -963,23 +970,20 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } catch (RuntimeException e) { this.logger.error("Transaction rolled back", e); + AfterRollbackProcessor afterRollbackProcessorToUse = + (AfterRollbackProcessor) getAfterRollbackProcessor(); if (recordList == null) { - getAfterRollbackProcessor().process(createRecordList(records), this.consumer, e, false); + afterRollbackProcessorToUse.process(createRecordList(records), this.consumer, e, false); } else { - getAfterRollbackProcessor().process(recordList, this.consumer, e, false); + afterRollbackProcessorToUse.process(recordList, this.consumer, e, false); } } } private List> createRecordList(final ConsumerRecords records) { - List> recordList; - recordList = new LinkedList>(); - Iterator> iterator = records.iterator(); - while (iterator.hasNext()) { - recordList.add(iterator.next()); - } - return recordList; + return StreamSupport.stream(records.spliterator(), false) + .collect(Collectors.toList()); } /** @@ -999,9 +1003,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } catch (RuntimeException e) { if (this.containerProperties.isAckOnError() && !this.autoCommit && producer == null) { - for (ConsumerRecord record : getHighestOffsetRecords(records)) { - this.acks.add(record); - } + this.acks.addAll(getHighestOffsetRecords(records)); } if (this.batchErrorHandler == null) { throw e; @@ -1073,17 +1075,14 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener @SuppressWarnings(RAW_TYPES) @Nullable Producer producer, RuntimeException e) { if (this.batchErrorHandler instanceof ContainerAwareBatchErrorHandler) { - ((ContainerAwareBatchErrorHandler) this.batchErrorHandler) - .handle(e, records, this.consumer, KafkaMessageListenerContainer.this.container); + this.batchErrorHandler.handle(e, records, this.consumer, KafkaMessageListenerContainer.this.container); } else { this.batchErrorHandler.handle(e, records, this.consumer); } // if the handler handled the error (no exception), go ahead and commit if (producer != null) { - for (ConsumerRecord record : getHighestOffsetRecords(records)) { - this.acks.add(record); - } + this.acks.addAll(getHighestOffsetRecords(records)); sendOffsetsToTransaction(producer); } } @@ -1101,7 +1100,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener * Invoke the listener with each record in a separate transaction. * @param records the records. */ - @SuppressWarnings({ RAW_TYPES }) + @SuppressWarnings({ "unchecked", RAW_TYPES }) private void invokeRecordListenerInTx(final ConsumerRecords records) { Iterator> iterator = records.iterator(); while (iterator.hasNext()) { @@ -1119,7 +1118,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener Producer producer = null; if (ListenerConsumer.this.kafkaTxManager != null) { producer = ((KafkaResourceHolder) TransactionSynchronizationManager - .getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory())) + .getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory())) .getProducer(); // NOSONAR } RuntimeException aborted = doInvokeRecordListener(record, producer, iterator); @@ -1137,7 +1136,8 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener while (iterator.hasNext()) { unprocessed.add(iterator.next()); } - getAfterRollbackProcessor().process(unprocessed, this.consumer, e, true); + ((AfterRollbackProcessor) getAfterRollbackProcessor()) + .process(unprocessed, this.consumer, e, true); } finally { TransactionSupport.clearTransactionIdSuffix(); @@ -1284,8 +1284,8 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener ackCurrent(record, null); } - public void ackCurrent(final ConsumerRecord record, @SuppressWarnings(RAW_TYPES) - @Nullable Producer producer) { + public void ackCurrent(final ConsumerRecord record, + @SuppressWarnings(RAW_TYPES) @Nullable Producer producer) { if (this.isRecordAck) { Map offsetsToCommit = @@ -1409,14 +1409,14 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener Map partitions = new HashMap<>(this.definedPartitions); Set beginnings = partitions.entrySet().stream() .filter(e -> SeekPosition.BEGINNING.equals(e.getValue().seekPosition)) - .map(e -> e.getKey()) + .map(Entry::getKey) .collect(Collectors.toSet()); - beginnings.forEach(k -> partitions.remove(k)); + beginnings.forEach(partitions::remove); Set ends = partitions.entrySet().stream() .filter(e -> SeekPosition.END.equals(e.getValue().seekPosition)) - .map(e -> e.getKey()) + .map(Entry::getKey) .collect(Collectors.toSet()); - ends.forEach(k -> partitions.remove(k)); + ends.forEach(partitions::remove); if (beginnings.size() > 0) { this.consumer.seekToBeginning(beginnings); } @@ -1432,7 +1432,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener if (offset < 0) { if (!metadata.relativeToCurrent) { - this.consumer.seekToEnd(Arrays.asList(topicPartition)); + this.consumer.seekToEnd(Collections.singletonList(topicPartition)); } newOffset = Math.max(0, this.consumer.position(topicPartition) + offset); } @@ -1674,21 +1674,24 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener ListenerConsumer.this.transactionTemplate .execute(new TransactionCallbackWithoutResult() { - @SuppressWarnings({ "unchecked", RAWTYPES }) - @Override - protected void doInTransactionWithoutResult(TransactionStatus status) { - KafkaResourceHolder holder = - (KafkaResourceHolder) TransactionSynchronizationManager - .getResource( - ListenerConsumer.this.kafkaTxManager.getProducerFactory()); - if (holder != null) { - holder.getProducer().sendOffsetsToTransaction( - Collections.singletonMap(partition, offsetAndMetadata), - ListenerConsumer.this.consumerGroupId); - } - } + @SuppressWarnings({"unchecked", RAWTYPES}) + @Override + protected void doInTransactionWithoutResult(TransactionStatus status) { + KafkaResourceHolder holder = + (KafkaResourceHolder) TransactionSynchronizationManager + .getResource( + ListenerConsumer.this.kafkaTxManager + .getProducerFactory()); + if (holder != null) { + holder.getProducer() + .sendOffsetsToTransaction( + Collections.singletonMap(partition, + offsetAndMetadata), + ListenerConsumer.this.consumerGroupId); + } + } - }); + }); }); } finally { 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 e095b555..d48cb020 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 @@ -1,5 +1,5 @@ /* - * Copyright 2016-2018 the original author or authors. + * Copyright 2016-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -183,7 +183,7 @@ public class EnableKafkaIntegrationTests { private RecordPassAllFilter recordFilter; @Autowired - private DefaultKafkaConsumerFactory consumerFactory; + private DefaultKafkaConsumerFactory consumerFactory; @Autowired private AtomicReference> consumerRef; @@ -873,7 +873,7 @@ public class EnableKafkaIntegrationTests { new ConcurrentKafkaListenerContainerFactory<>(); ConsumerFactory spiedCf = mock(ConsumerFactory.class); willAnswer(i -> { - Consumer spy = + Consumer spy = spy(consumerFactory().createConsumer(i.getArgument(0), i.getArgument(1), i.getArgument(2))); willAnswer(invocation -> { @@ -978,7 +978,7 @@ public class EnableKafkaIntegrationTests { } @Bean - public DefaultKafkaConsumerFactory consumerFactory() { + public DefaultKafkaConsumerFactory consumerFactory() { return new DefaultKafkaConsumerFactory<>(consumerConfigs()); } @@ -1840,12 +1840,12 @@ public class EnableKafkaIntegrationTests { } - public static class RecordPassAllFilter implements RecordFilterStrategy { + public static class RecordPassAllFilter implements RecordFilterStrategy { private boolean called; @Override - public boolean filter(ConsumerRecord consumerRecord) { + public boolean filter(ConsumerRecord consumerRecord) { called = true; return false; }