1

Friends, the following code runs without issue. However, although it takes less than a second to produce 1-million messages, it takes about two (qty. 2) minutes for producer closure (producer.close()) to complete (it blocks). I'm running a 3-Broker Confluent Kafka cluster on a stout Fedora PC, and hardware performance doesn't appear to be the issue.

I'm wondering if there's a Kafka configuration causing this. I have the following simple topic-partition schema:

Topic: myTopic Partition: 0 Leader: 0 Replicas: 0 Isr: 0

and min.insync.replicas=1

See below. Any ideas? Thank you in advance!

package com.example.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.clients.producer.ProducerConfig;
import java.util.Properties;

public class MyProducer {
    public static final Logger logger = LogManager.getLogger();

    public static void main (String[] args) {
        logger.info("Creating Kafka Producer...");

        Properties props = new Properties();
        props.put(ProducerConfig.CLIENT_ID_CONFIG, AppConfigs.applicationID);
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, AppConfigs.bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        KafkaProducer<Integer, String> producer = new KafkaProducer<Integer, String>(props);

        // The following loop takes less than a second to complete.
        logger.info("Start ...");
        for(int i = 0; i < AppConfigs.numEvents; i++) {
            producer.send(new ProducerRecord<>(AppConfigs.topicName, i, "msg" + i));
        }
        logger.info("End. Closing producer.");

        producer.close(); // However this blocks.
        System.out.println("I am here!"); // Getting here takes about ~2 minutes.
    }
}

1 Answer 1

2

doc

The producer consists of a pool of buffer space that holds records that haven't yet been transmitted to the server as well as a background I/O thread that is responsible for turning these records into requests and transmitting them to the cluster. Failure to close the producer after use will leak these resources.

The send() method is asynchronous. When called it adds the record to a buffer of pending record sends and immediately returns. This allows the producer to batch together individual records for efficiency.

This means that until the buffer is full, you will send messages from the application without any delay (asynchronously). But on closing, you will wait for all messages to be successfully delivered to the server.

And min.insync.replicas=X allows acks=all requests to continue to work when at least x replicas of the partition are in sync. Here, we saw an example with two replicas.

You can put ask = 0, this will allow you to reach the maximum throughput, since there will be a limitation only in the capacity of the network, but if there is a failure, messages may be lost. You always have to make compromises)

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.