GH-699: Fix Batch Listener with Kotlin

Fixes #699

showing issue with batch listener, along with fix

fixing checkstyle

fixing illegal import

Author: zach-schoenberger

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java

@@ -427,7 +427,13 @@ else if (parameterizedType.getRawType().equals(List.class)
 							Type paramType = parameterizedType.getActualTypeArguments()[0];
 							this.isConsumerRecordList =	paramType.equals(ConsumerRecord.class)
 									|| (paramType instanceof ParameterizedType
-										&& ((ParameterizedType) paramType).getRawType().equals(ConsumerRecord.class));
+										&& ((ParameterizedType) paramType).getRawType().equals(ConsumerRecord.class)
+									|| (paramType instanceof WildcardType
+										&& ((WildcardType) paramType).getUpperBounds() != null
+										&& ((WildcardType) paramType).getUpperBounds().length > 0
+										&& ((WildcardType) paramType).getUpperBounds()[0] instanceof ParameterizedType
+										&& ((ParameterizedType) ((WildcardType) paramType).getUpperBounds()[0]).getRawType().equals(ConsumerRecord.class))
+							);
 							boolean messageHasGeneric = paramType instanceof ParameterizedType
 									&& ((ParameterizedType) paramType).getRawType().equals(Message.class);
 							this.isMessageList = paramType.equals(Message.class) || messageHasGeneric;

spring-kafka/src/test/kotlin/org/springframework/kafka/listener/EnableKafkaKotlinTests.kt

@@ -15,6 +15,7 @@
  */
 
 import org.apache.kafka.clients.consumer.ConsumerConfig
+import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.apache.kafka.clients.producer.ProducerConfig
 import org.apache.kafka.common.serialization.StringDeserializer
 import org.apache.kafka.common.serialization.StringSerializer
@@ -59,13 +60,21 @@ class EnableKafkaKotlinTests {
 		assertThat(this.config.received).isEqualTo("foo")
 	}
 
+	@Test
+	fun `test batch listener`() {
+		this.template.send("kotlinTestTopic", "foo")
+		assertThat(this.config.latch.await(10, TimeUnit.SECONDS)).isTrue()
+		assertThat(this.config.batchReceived).isEqualTo("foo")
+	}
+
 	@Configuration
 	@EnableKafka
 	class Config {
 
 		lateinit var received: String
+		lateinit var batchReceived: String
 
-		val latch = CountDownLatch(1)
+		val latch = CountDownLatch(2)
 
 		@Value("\${" + KafkaEmbedded.SPRING_EMBEDDED_KAFKA_BROKERS + "}")
 		private lateinit var brokerAddresses: String
@@ -82,7 +91,6 @@ class EnableKafkaKotlinTests {
 		@Bean
 		fun kcf(): ConsumerFactory<String, String> {
 			val configs = HashMap<String, Any>()
-			configs["foo"] = "bar"
 			configs[ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG] = this.brokerAddresses
 			configs[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java
 			configs[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java
@@ -104,12 +112,27 @@ class EnableKafkaKotlinTests {
 			return factory
 		}
 
-		@KafkaListener(id = "kotlin", topics = ["kotlinTestTopic"])
+		@Bean
+		fun kafkaBatchListenerContainerFactory(): ConcurrentKafkaListenerContainerFactory<String, String> {
+			val factory: ConcurrentKafkaListenerContainerFactory<String, String>
+					= ConcurrentKafkaListenerContainerFactory()
+			factory.isBatchListener = true
+			factory.consumerFactory = kcf()
+			return factory
+		}
+
+		@KafkaListener(id = "kotlin", topics = ["kotlinTestTopic"], containerFactory = "kafkaListenerContainerFactory")
 		fun listen(value: String) {
 			this.received = value
 			this.latch.countDown()
 		}
 
+		@KafkaListener(id = "kotlin-batch", topics = ["kotlinTestTopic"], containerFactory = "kafkaBatchListenerContainerFactory")
+		fun batchListen(values: List<ConsumerRecord<String, String>>) {
+			this.batchReceived = values.first().value()
+			this.latch.countDown()
+		}
+
 	}
 
 }