producer.java

import org.apache.kafka.clients.producer.*
import java.util.*
import org.apache.kafka.common.serialization.StringSerializer
import java.lang.Exception

fun main(args: Array<String>) {
    val props = Properties()
    props[ProducerConfig.BOOTSTRAP_SERVERS_CONFIG] = "localhost:9092"
    props[ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java.name
    props[ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java.name

    val producer = KafkaProducer<Int, String>(props)
    producer.send(ProducerRecord("users.notification",
            0, Calendar.getInstance().timeInMillis, null, "hello"), object : Callback {
        override fun onCompletion(metadata: RecordMetadata?, exception: Exception?) {
            println("Send Complete")
        }
    })

    Thread.sleep(120 * 1000)
}