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

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) {
 
 		}
+
 	}
+
 }

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<C extends AbstractMe
 
 	private GenericErrorHandler<?> errorHandler;
 
-	private ConsumerFactory<K, V> consumerFactory;
+	private ConsumerFactory<? super K, ? super V> consumerFactory;
 
 	private Boolean autoStartup;
 
@@ -69,7 +69,7 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 
 	private MessageConverter messageConverter;
 
-	private RecordFilterStrategy<K, V> recordFilterStrategy;
+	private RecordFilterStrategy<? super K, ? super V> recordFilterStrategy;
 
 	private Boolean ackDiscarded;
 
@@ -85,7 +85,7 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 
 	private KafkaTemplate<?, ?> replyTemplate;
 
-	private AfterRollbackProcessor<K, V> afterRollbackProcessor;
+	private AfterRollbackProcessor<? super K, ? super V> afterRollbackProcessor;
 
 	private ReplyHeadersConfigurer replyHeadersConfigurer;
 
@@ -93,11 +93,11 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 	 * Specify a {@link ConsumerFactory} to use.
 	 * @param consumerFactory The consumer factory.
 	 */
-	public void setConsumerFactory(ConsumerFactory<K, V> consumerFactory) {
+	public void setConsumerFactory(ConsumerFactory<? super K, ? super V> consumerFactory) {
 		this.consumerFactory = consumerFactory;
 	}
 
-	public ConsumerFactory<K, V> getConsumerFactory() {
+	public ConsumerFactory<? super K, ? super V> getConsumerFactory() {
 		return this.consumerFactory;
 	}
 
@@ -131,7 +131,7 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 	 * Set the record filter strategy.
 	 * @param recordFilterStrategy the strategy.
 	 */
-	public void setRecordFilterStrategy(RecordFilterStrategy<K, V> recordFilterStrategy) {
+	public void setRecordFilterStrategy(RecordFilterStrategy<? super K, ? super V> recordFilterStrategy) {
 		this.recordFilterStrategy = recordFilterStrategy;
 	}
 
@@ -231,7 +231,7 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 	 * @param afterRollbackProcessor the processor.
 	 * @since 1.3.5
 	 */
-	public void setAfterRollbackProcessor(AfterRollbackProcessor<K, V> afterRollbackProcessor) {
+	public void setAfterRollbackProcessor(AfterRollbackProcessor<? super K, ? super V> afterRollbackProcessor) {
 		this.afterRollbackProcessor = afterRollbackProcessor;
 	}
 
@@ -258,11 +258,13 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 		if (this.errorHandler != null) {
 			if (Boolean.TRUE.equals(this.batchListener)) {
 				Assert.state(this.errorHandler instanceof BatchErrorHandler,
-						"The error handler must be a BatchErrorHandler, not " + this.errorHandler.getClass().getName());
+						() -> "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());
 			}
 		}
 	}

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<K, V>
 		return this.replyTemplate;
 	}
 
-	protected RecordFilterStrategy<K, V> getRecordFilterStrategy() {
+	protected RecordFilterStrategy<? super K, ? super V> getRecordFilterStrategy() {
 		return this.recordFilterStrategy;
 	}
 
@@ -276,8 +276,9 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 	 * Set a {@link RecordFilterStrategy} implementation.
 	 * @param recordFilterStrategy the strategy implementation.
 	 */
-	public void setRecordFilterStrategy(RecordFilterStrategy<K, V> recordFilterStrategy) {
-		this.recordFilterStrategy = recordFilterStrategy;
+	@SuppressWarnings("unchecked")
+	public void setRecordFilterStrategy(RecordFilterStrategy<? super K, ? super V> recordFilterStrategy) {
+		this.recordFilterStrategy = (RecordFilterStrategy<K, V>) recordFilterStrategy;
 	}
 
 	protected boolean isAckDiscarded() {
@@ -285,8 +286,7 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 	}
 
 	/**
-	 * 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<K, V>
 	}
 
 	/**
-	 * 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<? extends Object> recoveryCallback) {

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<K, V>
 
 	private int phase = DEFAULT_PHASE;
 
-	private AfterRollbackProcessor<K, V> afterRollbackProcessor = new DefaultAfterRollbackProcessor<>();
+	private AfterRollbackProcessor<? super K, ? super V> afterRollbackProcessor =
+			new DefaultAfterRollbackProcessor<>();
 
 	private volatile boolean running = false;
 
@@ -98,11 +100,12 @@ public abstract class AbstractMessageListenerContainer<K, V>
 	 * @param consumerFactory the factory.
 	 * @param containerProperties the properties.
 	 */
-	protected AbstractMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
+	@SuppressWarnings("unchecked")
+	protected AbstractMessageListenerContainer(ConsumerFactory<? super K, ? super V> consumerFactory,
 			ContainerProperties containerProperties) {
 
 		Assert.notNull(containerProperties, "'containerProperties' cannot be null");
-		this.consumerFactory = consumerFactory;
+		this.consumerFactory = (ConsumerFactory<K, V>) consumerFactory;
 		if (containerProperties.getTopics() != null) {
 			this.containerProperties = new ContainerProperties(containerProperties.getTopics());
 		}
@@ -221,7 +224,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
 		return this.phase;
 	}
 
-	protected AfterRollbackProcessor<K, V> getAfterRollbackProcessor() {
+	protected AfterRollbackProcessor<? super K, ? super V> getAfterRollbackProcessor() {
 		return this.afterRollbackProcessor;
 	}
 
@@ -232,7 +235,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
 	 * @param afterRollbackProcessor the processor.
 	 * @since 1.3.5
 	 */
-	public void setAfterRollbackProcessor(AfterRollbackProcessor<K, V> afterRollbackProcessor) {
+	public void setAfterRollbackProcessor(AfterRollbackProcessor<? super K, ? super V> afterRollbackProcessor) {
 		Assert.notNull(afterRollbackProcessor, "'afterRollbackProcessor' cannot be null");
 		this.afterRollbackProcessor = afterRollbackProcessor;
 	}
@@ -252,9 +255,8 @@ public abstract class AbstractMessageListenerContainer<K, V>
 		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<K, V>
 
 	protected void checkTopics() {
 		if (this.containerProperties.isMissingTopicsFatal() && this.containerProperties.getTopicPattern() == null) {
-			try (Consumer<K, V> consumer = this.consumerFactory.createConsumer(this.containerProperties.getGroupId(),
-					this.containerProperties.getClientId(), null)) {
+			try (Consumer<K, V> 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<String> missing = new ArrayList<>();
@@ -293,12 +296,12 @@ public abstract class AbstractMessageListenerContainer<K, V>
 			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<K, V>
 		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<K, V>
 	}
 
 	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
 	 */

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<K, V> extends AbstractMessageLis
 	 * @param consumerFactory the consumer factory.
 	 * @param containerProperties the container properties.
 	 */
-	public ConcurrentMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
+	public ConcurrentMessageListenerContainer(ConsumerFactory<? super K, ? super V> consumerFactory,
 			ContainerProperties containerProperties) {
+
 		super(consumerFactory, containerProperties);
 		Assert.notNull(consumerFactory, "A ConsumerFactory must be provided");
 	}
@@ -100,7 +101,7 @@ public class ConcurrentMessageListenerContainer<K, V> 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<K, V> 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<K, V> extends AbstractMessageLis
 			for (int i = 0; i < this.concurrency; i++) {
 				KafkaMessageListenerContainer<K, V> 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<K, V> extends AbstractMessageLis
 			}
 			for (KafkaMessageListenerContainer<K, V> 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<K, V> 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

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<K, V> extends AbstractMessageListenerContainer<K, V> { // NOSONAR comment density
+public class KafkaMessageListenerContainer<K, V>
+		extends AbstractMessageListenerContainer<K, V> { // NOSONAR comment density
 
 	private static final int DEFAULT_ACK_TIME = 5000;
 
@@ -135,8 +136,9 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 	 * @param consumerFactory the consumer factory.
 	 * @param containerProperties the container properties.
 	 */
-	public KafkaMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
+	public KafkaMessageListenerContainer(ConsumerFactory<? super K, ? super V> consumerFactory,
 			ContainerProperties containerProperties) {
+
 		this(null, consumerFactory, containerProperties, (TopicPartitionInitialOffset[]) null);
 	}
 
@@ -147,8 +149,9 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 	 * @param containerProperties the container properties.
 	 * @param topicPartitions the topics/partitions; duplicates are eliminated.
 	 */
-	public KafkaMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
+	public KafkaMessageListenerContainer(ConsumerFactory<? super K, ? super V> consumerFactory,
 			ContainerProperties containerProperties, TopicPartitionInitialOffset... topicPartitions) {
+
 		this(null, consumerFactory, containerProperties, topicPartitions);
 	}
 
@@ -159,8 +162,9 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 	 * @param containerProperties the container properties.
 	 */
 	KafkaMessageListenerContainer(AbstractMessageListenerContainer<K, V> container,
-			ConsumerFactory<K, V> consumerFactory,
+			ConsumerFactory<? super K, ? super V> consumerFactory,
 			ContainerProperties containerProperties) {
+
 		this(container, consumerFactory, containerProperties, (TopicPartitionInitialOffset[]) null);
 	}
 
@@ -173,8 +177,9 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 	 * @param topicPartitions the topics/partitions; duplicates are eliminated.
 	 */
 	KafkaMessageListenerContainer(AbstractMessageListenerContainer<K, V> container,
-			ConsumerFactory<K, V> consumerFactory,
+			ConsumerFactory<? super K, ? super V> 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<K, V> 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<K, V> 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<K, V> 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<K, V> 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<String, Object> props = KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties();
+			Map<String, Object> props =
+					KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties();
 			this.checkNullKeyForExceptions = checkDeserializer(findDeserializerClass(props, false));
 			this.checkNullValueForExceptions = checkDeserializer(findDeserializerClass(props, true));
 		}
@@ -564,7 +573,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 			}
 		}
 
-		private void subscribeOrAssignTopics(final Consumer<K, V> consumer) {
+		private void subscribeOrAssignTopics(final Consumer<? super K, ? super V> consumer) {
 			if (KafkaMessageListenerContainer.this.topicPartitions == null) {
 				ConsumerRebalanceListener rebalanceListener = new ListenerConsumerRebalanceListener();
 				if (this.containerProperties.getTopicPattern() != null) {
@@ -590,9 +599,7 @@ public class KafkaMessageListenerContainer<K, V> 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<K, V> extends AbstractMessageListener
 			}
 		}
 
-		@SuppressWarnings({ RAW_TYPES })
+		@SuppressWarnings({ "unchecked", RAW_TYPES })
 		private void invokeBatchListenerInTx(final ConsumerRecords<K, V> records,
 				final List<ConsumerRecord<K, V>> recordList) {
 			try {
@@ -952,7 +959,7 @@ public class KafkaMessageListenerContainer<K, V> 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<K, V> extends AbstractMessageListener
 			}
 			catch (RuntimeException e) {
 				this.logger.error("Transaction rolled back", e);
+				AfterRollbackProcessor<K, V> afterRollbackProcessorToUse =
+						(AfterRollbackProcessor<K, V>) 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<ConsumerRecord<K, V>> createRecordList(final ConsumerRecords<K, V> records) {
-			List<ConsumerRecord<K, V>> recordList;
-			recordList = new LinkedList<ConsumerRecord<K, V>>();
-			Iterator<ConsumerRecord<K, V>> 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<K, V> extends AbstractMessageListener
 			}
 			catch (RuntimeException e) {
 				if (this.containerProperties.isAckOnError() && !this.autoCommit && producer == null) {
-					for (ConsumerRecord<K, V> 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<K, V> 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<K, V> record : getHighestOffsetRecords(records)) {
-					this.acks.add(record);
-				}
+				this.acks.addAll(getHighestOffsetRecords(records));
 				sendOffsetsToTransaction(producer);
 			}
 		}
@@ -1101,7 +1100,7 @@ public class KafkaMessageListenerContainer<K, V> 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<K, V> records) {
 			Iterator<ConsumerRecord<K, V>> iterator = records.iterator();
 			while (iterator.hasNext()) {
@@ -1119,7 +1118,7 @@ public class KafkaMessageListenerContainer<K, V> 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<K, V> extends AbstractMessageListener
 					while (iterator.hasNext()) {
 						unprocessed.add(iterator.next());
 					}
-					getAfterRollbackProcessor().process(unprocessed, this.consumer, e, true);
+					((AfterRollbackProcessor<K, V>) getAfterRollbackProcessor())
+							.process(unprocessed, this.consumer, e, true);
 				}
 				finally {
 					TransactionSupport.clearTransactionIdSuffix();
@@ -1284,8 +1284,8 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 			ackCurrent(record, null);
 		}
 
-		public void ackCurrent(final ConsumerRecord<K, V> record, @SuppressWarnings(RAW_TYPES)
-				@Nullable Producer producer) {
+		public void ackCurrent(final ConsumerRecord<K, V> record,
+				@SuppressWarnings(RAW_TYPES) @Nullable Producer producer) {
 
 			if (this.isRecordAck) {
 				Map<TopicPartition, OffsetAndMetadata> offsetsToCommit =
@@ -1409,14 +1409,14 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 			Map<TopicPartition, OffsetMetadata> partitions = new HashMap<>(this.definedPartitions);
 			Set<TopicPartition> 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<TopicPartition> 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<K, V> 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<K, V> 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 {

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<Integer, String> consumerFactory;
+	private DefaultKafkaConsumerFactory<Integer, CharSequence> consumerFactory;
 
 	@Autowired
 	private AtomicReference<Consumer<?, ?>> consumerRef;
@@ -873,7 +873,7 @@ public class EnableKafkaIntegrationTests {
 					new ConcurrentKafkaListenerContainerFactory<>();
 			ConsumerFactory spiedCf = mock(ConsumerFactory.class);
 			willAnswer(i -> {
-				Consumer<Integer, String> spy =
+				Consumer<Integer, CharSequence> 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<Integer, String> consumerFactory() {
+		public DefaultKafkaConsumerFactory<Integer, CharSequence> consumerFactory() {
 			return new DefaultKafkaConsumerFactory<>(consumerConfigs());
 		}
 
@@ -1840,12 +1840,12 @@ public class EnableKafkaIntegrationTests {
 
 	}
 
-	public static class RecordPassAllFilter implements RecordFilterStrategy<Integer, String> {
+	public static class RecordPassAllFilter implements RecordFilterStrategy<Integer, CharSequence> {
 
 		private boolean called;
 
 		@Override
-		public boolean filter(ConsumerRecord<Integer, String> consumerRecord) {
+		public boolean filter(ConsumerRecord<Integer, CharSequence> consumerRecord) {
 			called = true;
 			return false;
 		}