I am sending some Kafka messages over the Internet. The message sizes
are about 400K. The essential logic of my code is as follows:

Properties config = new Properties();
config.put("bootstrap.servers", "...");
config.put("client.id", "...");
config.put("key.serializer",
"org.apache.kafka.common.serialization.ByteArraySerializer");
config.put("value.serializer",
"org.apache.kafka.common.serialization.ByteArraySerializer");
config.put("transactional.id", "...");

producer = new KafkaProducer<>(config());
producer.initTransactions();

producer.beginTransaction();
for (byte[] bytesMessage : batch) {
  producer.send(
      new ProducerRecord<>("...", null, bytesMessage));
}
producer.commitTransaction();

I found that there were two records for some bytesMessage on the
topic. Is there something wrong with my code? Or duplicated message
deliveries are still possible with transactional.id set.

-- 
Jingguo

Reply via email to