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.
}
}