Improve container test coverage

From 78.5% to 83.8% and fix bugs found:

- Deprecate unused ctor
- Fix check for proper error handler for listener type

* More coverage and polishing.

Author: garyrussell

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

@@ -271,7 +271,7 @@ public final void start() {
 		checkGroupId();
 		synchronized (this.lifecycleMonitor) {
 			if (!isRunning()) {
-				Assert.isTrue(this.containerProperties.getMessageListener() instanceof GenericMessageListener,
+				Assert.state(this.containerProperties.getMessageListener() instanceof GenericMessageListener,
 						() -> "A " + GenericMessageListener.class.getName() + " implementation must be provided");
 				doStart();
 			}

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

@@ -16,7 +16,6 @@
 
 package org.springframework.kafka.listener;
 
-import java.lang.reflect.Type;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -145,7 +144,10 @@ public KafkaMessageListenerContainer(ConsumerFactory<? super K, ? super V> consu
 	 * @param consumerFactory the consumer factory.
 	 * @param containerProperties the container properties.
 	 * @param topicPartitions the topics/partitions; duplicates are eliminated.
+	 * @deprecated - the topicPartitions should be provided in the
+	 * {@link ContainerProperties}.
 	 */
+	@Deprecated
 	public KafkaMessageListenerContainer(ConsumerFactory<? super K, ? super V> consumerFactory,
 			ContainerProperties containerProperties, TopicPartitionInitialOffset... topicPartitions) {
 
@@ -237,7 +239,7 @@ else if (partitionsListenerConsumer.assignedPartitions != null) {
 
 	@Override
 	public boolean isContainerPaused() {
-		return isPaused() && this.listenerConsumer.consumerPaused;
+		return isPaused() && this.listenerConsumer != null && this.listenerConsumer.consumerPaused;
 	}
 
 	@Override
@@ -266,13 +268,11 @@ protected void doStart() {
 		checkAckMode(containerProperties);
 
 		Object messageListener = containerProperties.getMessageListener();
-		Assert.state(messageListener != null, "A MessageListener is required");
 		if (containerProperties.getConsumerTaskExecutor() == null) {
 			SimpleAsyncTaskExecutor consumerExecutor = new SimpleAsyncTaskExecutor(
 					(getBeanName() == null ? "" : getBeanName()) + "-C-");
 			containerProperties.setConsumerTaskExecutor(consumerExecutor);
 		}
-		Assert.state(messageListener instanceof GenericMessageListener, "Listener must be a GenericListener");
 		GenericMessageListener<?> listener = (GenericMessageListener<?>) messageListener;
 		ListenerType listenerType = determineListenerType(listener);
 		this.listenerConsumer = new ListenerConsumer(listener, listenerType);
@@ -480,7 +480,7 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private boolean taskSchedulerExplicitlySet;
 
-		private boolean consumerPaused;
+		volatile boolean consumerPaused;
 
 		private long lastReceive = System.currentTimeMillis();
 
@@ -546,12 +546,11 @@ else if (listener instanceof MessageListener) {
 				this.taskScheduler = threadPoolTaskScheduler;
 			}
 			this.monitorTask = this.taskScheduler.scheduleAtFixedRate(this::checkConsumer,
-					this.containerProperties.getMonitorInterval() * 1000); // NOSONAR magic #
+					Duration.ofSeconds(this.containerProperties.getMonitorInterval()));
 			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));
 			this.syncCommitTimeout = determineSyncCommitTimeout();
@@ -642,7 +641,8 @@ private void subscribeOrAssignTopics(final Consumer<? super K, ? super V> subscr
 					subscribingConsumer.subscribe(this.containerProperties.getTopicPattern(), rebalanceListener);
 				}
 				else {
-					subscribingConsumer.subscribe(Arrays.asList(this.containerProperties.getTopics()), rebalanceListener);
+					subscribingConsumer.subscribe(Arrays.asList(this.containerProperties.getTopics()),
+							rebalanceListener);
 				}
 			}
 			else {
@@ -717,22 +717,14 @@ public void seekToEnd(String topic, int partition) {
 
 		private void validateErrorHandler(boolean batch) {
 			GenericErrorHandler<?> errHandler = KafkaMessageListenerContainer.this.getGenericErrorHandler();
-			if (this.errorHandler == null) {
+			if (errHandler == null) {
 				return;
 			}
-			Type[] genericInterfaces = errHandler.getClass().getGenericInterfaces();
-			boolean ok = false;
-			for (Type t : genericInterfaces) {
-				if (t.equals(ErrorHandler.class)) {
-					ok = !batch;
-					break;
-				}
-				else if (t.equals(BatchErrorHandler.class)) {
-					ok = batch;
-					break;
-				}
-			}
-			Assert.state(ok, "Error handler is not compatible with the message listener, expecting an instance of "
+			Class<?> clazz = errHandler.getClass();
+			Assert.state(batch
+						? BatchErrorHandler.class.isAssignableFrom(clazz)
+						: ErrorHandler.class.isAssignableFrom(clazz),
+					"Error handler is not compatible with the message listener, expecting an instance of "
 					+ (batch ? "BatchErrorHandler" : "ErrorHandler") + " not " + errHandler.getClass().getName());
 		}
 
@@ -821,7 +813,7 @@ private void debugRecords(ConsumerRecords<K, V> records) {
 							.flatMap(p -> records.records(p).stream())
 							// map to same format as send metadata toString()
 							.map(r -> r.topic() + "-" + r.partition() + "@" + r.offset())
-							.collect(Collectors.toList()));
+							.collect(Collectors.toList()).toString());
 				}
 			}
 		}
@@ -1022,7 +1014,7 @@ public void doInTransactionWithoutResult(TransactionStatus s) {
 						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) {
@@ -1097,7 +1089,7 @@ private RuntimeException doInvokeBatchListener(final ConsumerRecords<K, V> recor
 					this.logger.error("Error handler threw an exception", ee);
 					return ee;
 				}
-				catch (Error er) { //NOSONAR
+				catch (Error er) { // NOSONAR
 					this.logger.error("Error handler threw an error", er);
 					throw er;
 				}
@@ -1114,7 +1106,8 @@ private void invokeBatchOnMessage(final ConsumerRecords<K, V> records, List<Cons
 				this.batchListener.onMessage(records,
 						this.isAnyManualAck
 								? new ConsumerBatchAcknowledgment(records)
-								: null, this.consumer);
+								: null,
+						this.consumer);
 			}
 			else {
 				doInvokeBatchOnMessage(records, recordList);
@@ -1202,7 +1195,7 @@ public void doInTransactionWithoutResult(TransactionStatus s) {
 							if (ListenerConsumer.this.kafkaTxManager != null) {
 								producer = ((KafkaResourceHolder) TransactionSynchronizationManager
 										.getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory()))
-										.getProducer(); // NOSONAR
+												.getProducer(); // NOSONAR
 							}
 							RuntimeException aborted = doInvokeRecordListener(record, producer, iterator);
 							if (aborted != null) {
@@ -1290,7 +1283,7 @@ private RuntimeException doInvokeRecordListener(final ConsumerRecord<K, V> recor
 					this.logger.error("Error handler threw an exception", ee);
 					return ee;
 				}
-				catch (Error er) { //NOSONAR
+				catch (Error er) { // NOSONAR
 					this.logger.error("Error handler threw an error", er);
 					throw er;
 				}
@@ -1777,7 +1770,7 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 								ListenerConsumer.this.transactionTemplate
 										.execute(new TransactionCallbackWithoutResult() {
 
-											@SuppressWarnings({UNCHECKED, RAWTYPES})
+											@SuppressWarnings({ UNCHECKED, RAWTYPES })
 											@Override
 											protected void doInTransactionWithoutResult(TransactionStatus status) {
 												KafkaResourceHolder holder =

spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingDeserializerTests.java

@@ -171,7 +171,7 @@ else if (r.key() == null && t.getCause() instanceof DeserializationException) {
 		public ConsumerFactory<String, String> cf() {
 			Map<String, Object> props = KafkaTestUtils.consumerProps(TOPIC + ".g1", "false", embeddedKafka());
 			props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
-			props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
+			props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class.getName());
 			props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
 			props.put(ErrorHandlingDeserializer2.KEY_DESERIALIZER_CLASS, FailSometimesDeserializer.class);
 			props.put(ErrorHandlingDeserializer2.VALUE_DESERIALIZER_CLASS, FailSometimesDeserializer.class.getName());