Java/Scala Kafka Producer does not send message to topic

Question

I'm having a problem with sending a serialized XML to my Kafka topic. Whenever I run my code, I don't get any exceptions or error message, but still I can't see any of my messages in the Kafka-topic.

My Kafka-Producer settings are:

def WartungsdbKafkaConnector(args: Array[String]): Unit = {  
    val xmlFile = args(0)
    val record = getRecord(xmlFile)
    val kafkaProducer = getKafkaProducer
    kafkaProducer.send(record)
}

protected def getRecord(xmlFile: String): ProducerRecord[String, String] = {
    val lines = scala.io.Source.fromFile(xmlFile).mkString
    val xml = scala.xml.XML.loadString(lines)
    val paramPress = xml \ "PARAMETER" \ "PRESS"
    val databaseId = allCatch.opt {paramPress.\@("NUMBER")}
    val key = databaseId.get
    val topic = args(1)
    new ProducerRecord(topic, key, lines)
}


protected def getKafkaProducer: KafkaProducer[String, String] = {  
    val props = new Properties
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 
    "ec-x.eu-west-1.compute.amazonaws.com:9092," +
    "ec2-x.eu-west-1.compute.amazonaws.com:9092," +
    "ec2-x.eu-west-1.compute.amazonaws.com:9092")
    props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true")
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName)
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName)
    props.put(ProducerConfig.LINGER_MS_CONFIG, "100")
    props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy")
    props.put(ProducerConfig.RETRIES_CONFIG, "20")
    props.put(ProducerConfig.ACKS_CONFIG, "all")

    new KafkaProducer[String, String](props)
}

When I run the code, I get:

[main] INFO org.apache.kafka.clients.producer.ProducerConfig - ProducerConfig values:
    acks = all
    batch.size = 16384
    bootstrap.servers = [ec2-x.eu-west-1.compute.amazonaws.com:9092, 
    ec2-x.eu-west-1.compute.amazonaws.com:9092, 
    ec2-x.eu-west-1.compute.amazonaws.com:9092]
    buffer.memory = 33554432
    client.id =
    compression.type = snappy
    connections.max.idle.ms = 540000
    enable.idempotence = true
    interceptor.classes = []
    key.serializer = class org.apache.kafka.common.serialization.StringSerializer
    linger.ms = 100
    max.block.ms = 60000
    max.in.flight.requests.per.connection = 5
    max.request.size = 1048576
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
    receive.buffer.bytes = 32768
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    request.timeout.ms = 30000
    retries = 2
    retry.backoff.ms = 100
    sasl.client.callback.handler.class = null
    sasl.jaas.config = null
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.login.callback.handler.class = null
    sasl.login.class = null
    sasl.login.refresh.buffer.seconds = 300
    sasl.login.refresh.min.period.seconds = 60
    sasl.login.refresh.window.factor = 0.8
    sasl.login.refresh.window.jitter = 0.05
    sasl.mechanism = GSSAPI
    security.protocol = PLAINTEXT
    send.buffer.bytes = 131072
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
    ssl.endpoint.identification.algorithm = https
    ssl.key.password = null
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.location = null
    ssl.keystore.password = null
    ssl.keystore.type = JKS
    ssl.protocol = TLS
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.location = null
    ssl.truststore.password = null
    ssl.truststore.type = JKS
    transaction.timeout.ms = 60000
    transactional.id = null
    value.serializer = class org.apache.kafka.common.serialization.StringSerializer

[main] INFO org.apache.kafka.clients.producer.KafkaProducer - [Producer 
clientId=producer-1] Instantiated an idempotent producer.
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version : 
2.0.0
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId : 
3402a8361b734732
[kafka-producer-network-thread | producer-1] INFO 
org.apache.kafka.clients.Metadata - Cluster ID: xeb6oWNpTgSQ_9FHctZ2ng
[kafka-producer-network-thread | producer-1] INFO 
org.apache.kafka.clients.producer.internals.TransactionManager - [Producer 
clientId=producer-1] ProducerId set to 150671 with epoch 0

Any Idea how to make it work? Thanks in advance!

Answer 1

You're not flushing, waiting for, or closing the producer, so the app just stops without sending data.

Producers batch data for a configurable amount of time and messages to reduce the number of send requests actually get to the brokers.

Try

kafkaProducer.send(record)  // optionally call get() on this to capture the result and potential errors 
kafkaProducer.flush() 
kafkaProducer.close()

Most importantly, never forget to close the producer (or a consumer)