Add Kotlin test case

* Use 'kotlin-spring' gradle plugin

build.gradle

@@ -1,10 +1,13 @@
 buildscript {
+	ext.kotlinVersion = '1.2.41'
 	repositories {
 		maven { url 'https://repo.spring.io/plugins-release' }
 	}
 	dependencies {
 		classpath 'io.spring.gradle:docbook-reference-plugin:0.3.1'
 		classpath 'org.asciidoctor:asciidoctor-gradle-plugin:1.5.3'
+		classpath "org.jetbrains.kotlin:kotlin-gradle-plugin:$kotlinVersion"
+		classpath "org.jetbrains.kotlin:kotlin-allopen:$kotlinVersion"
 	}
 }
 
@@ -49,14 +52,23 @@ subprojects { subproject ->
 	apply plugin: 'idea'
 	apply plugin: 'jacoco'
 	apply plugin: 'checkstyle'
+	apply plugin: 'kotlin'
+	apply plugin: 'kotlin-spring'
 
 	compileJava {
 		sourceCompatibility = 1.8
 		targetCompatibility = 1.8
 	}
 
+	compileTestKotlin {
+		kotlinOptions {
+			jvmTarget = '1.8'
+		}
+	}
+
 	ext {
 		assertjVersion = '3.9.1'
+		assertkVersion = '0.10'
 		hamcrestVersion = '1.3'
 		jacksonVersion = '2.9.5'
 		jaywayJsonPathVersion = '2.4.0'
@@ -94,6 +106,13 @@ subprojects { subproject ->
 		testCompileOnly 'org.apiguardian:apiguardian-api:1.0.0'
 
 		testRuntime "org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion"
+
+		testCompile("com.willowtreeapps.assertk:assertk:$assertkVersion") {
+				exclude group: 'org.jetbrains.kotlin', module: 'kotlin-reflect'
+		}
+
+		testCompile "org.jetbrains.kotlin:kotlin-stdlib:$kotlinVersion"
+		testRuntime "org.jetbrains.kotlin:kotlin-reflect:$kotlinVersion"
 	}
 
 	// enable all compiler warnings; individual projects may customize further

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

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2016-2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import org.apache.kafka.clients.consumer.ConsumerConfig
+import org.apache.kafka.clients.producer.ProducerConfig
+import org.apache.kafka.common.serialization.StringDeserializer
+import org.apache.kafka.common.serialization.StringSerializer
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.Test
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.kafka.annotation.EnableKafka
+import org.springframework.kafka.annotation.KafkaListener
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory
+import org.springframework.kafka.core.*
+import org.springframework.kafka.test.context.EmbeddedKafka
+import org.springframework.kafka.test.rule.KafkaEmbedded
+import org.springframework.test.annotation.DirtiesContext
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+
+/**
+ * @author Gary Russell
+ * @since 2.2
+ */
+
+@SpringJUnitConfig
+@DirtiesContext
+@EmbeddedKafka(topics = ["kotlinTestTopic"])
+class EnableKafkaKotlinTests {
+
+	@Autowired
+	private lateinit var config: Config
+
+	@Autowired
+	private lateinit var template: KafkaTemplate<String, String>
+
+	@Test
+	fun `test listener`() {
+		this.template.send("kotlinTestTopic", "foo")
+		assertThat(this.config.latch.await(10, TimeUnit.SECONDS)).isTrue()
+		assertThat(this.config.received).isEqualTo("foo")
+	}
+
+	@Configuration
+	@EnableKafka
+	class Config {
+
+		lateinit var received: String
+
+		val latch = CountDownLatch(1)
+
+		@Value("\${" + KafkaEmbedded.SPRING_EMBEDDED_KAFKA_BROKERS + "}")
+		private lateinit var brokerAddresses: String
+
+		@Bean
+		fun kpf(): ProducerFactory<String, String> {
+			val configs = HashMap<String, Any>()
+			configs[ProducerConfig.BOOTSTRAP_SERVERS_CONFIG] = this.brokerAddresses
+			configs[ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java
+			configs[ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java
+			return DefaultKafkaProducerFactory(configs)
+		}
+
+		@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
+			configs[ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG] = false
+			configs[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "earliest"
+			return DefaultKafkaConsumerFactory(configs)
+		}
+
+		@Bean
+		fun kt(): KafkaTemplate<String, String> {
+			return KafkaTemplate(kpf())
+		}
+
+		@Bean
+		fun kafkaListenerContainerFactory(): ConcurrentKafkaListenerContainerFactory<String, String> {
+			val factory: ConcurrentKafkaListenerContainerFactory<String, String>
+				= ConcurrentKafkaListenerContainerFactory()
+			factory.consumerFactory = kcf()
+			return factory
+		}
+
+		@KafkaListener(id = "kotlin", topics = ["kotlinTestTopic"])
+		fun listen(value: String) {
+			this.received = value
+			this.latch.countDown()
+		}
+
+	}
+
+}