Producing Messages

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);

props.put("acks", "all");
props.put("enable.idempotence", "true");
props.put("compression.type", "lz4");

props.put("bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092");

props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("schema.registry.url", "http://localhost:8081");

ProducerRecord<String, String> record =
    new ProducerRecord<>("orders", "order-99", "{\"amount\":42}");
producer.send(record);

try {
    RecordMetadata md = producer.send(record).get();
    System.out.printf("offset=%d%n", md.offset());
} catch (ExecutionException e) {
    log.error("send failed", e.getCause());
}

producer.send(record, (md, ex) -> {
    if (ex != null) log.error("failed", ex);
    else log.info("sent to {}-{}", md.topic(), md.partition());
});

producer.send(new ProducerRecord<>("accounts", accountId, balanceJson));

// topic, partition, key, value
producer.send(new ProducerRecord<>("orders", 3, "k", "v"));

ProducerRecord<String, String> r = new ProducerRecord<>("orders", "k", "v");
r.headers().add("trace-id", traceId.getBytes(StandardCharsets.UTF_8));
r.headers().add("source", "web".getBytes(StandardCharsets.UTF_8));
producer.send(r);

props.put("acks", "all");
props.put("min.insync.replicas", "2"); // enforced at broker/topic

for (Order o : batch) producer.send(toRecord(o));
producer.flush();   // wait for all acks
producer.close();