Producing

# order-42:hello kafka  (key:value, split by -K:)
echo "order-42:hello kafka" | kcat -b localhost:9092 -t orders -P -K: \
  -H "source=demo" \
  -X acks=all -X enable.idempotence=true -X linger.ms=50 -X batch.size=65536

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/twmb/franz-go/pkg/kgo"
)

func main() {
	cl, err := kgo.NewClient(
		kgo.SeedBrokers("localhost:9092"),
		kgo.RequiredAcks(kgo.AllISRAcks()), // acks=all
		kgo.ProducerLinger(50*time.Millisecond),
		kgo.ProducerBatchMaxBytes(64<<10),
		// Idempotent writes are ON by default; kgo.DisableIdempotentWrite() would turn them off.
	)
	if err != nil {
		panic(err)
	}
	defer cl.Close()

	record := &kgo.Record{
		Topic: "orders",
		Key:   []byte("order-42"),
		Value: []byte("hello kafka"),
		Headers: []kgo.RecordHeader{
			{Key: "source", Value: []byte("demo")},
		},
	}
	res := cl.ProduceSync(context.Background(), record)
	if err := res.FirstErr(); err != nil {
		panic(err)
	}
	r, _ := res.First()
	fmt.Printf("produced to partition=%d offset=%d\n", r.Partition, r.Offset)
}

from confluent_kafka import Producer

conf = {
    "bootstrap.servers": "localhost:9092",
    "acks": "all",
    "enable.idempotence": True,  # requires acks=all; the client enforces this
    "linger.ms": 50,
    "batch.size": 65536,
}
producer = Producer(conf)


def on_delivery(err, msg):
    if err is not None:
        raise RuntimeError(f"delivery failed: {err}")
    print(f"produced to partition={msg.partition()} offset={msg.offset()}")


producer.produce(
    topic="orders",
    key="order-42",
    value="hello kafka",
    headers=[("source", b"demo")],
    callback=on_delivery,
)
producer.flush(10)

import java.nio.charset.StandardCharsets;
import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

public final class Main {
    public static void main(String[] args) throws Exception {
        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");
        props.put("acks", "all");
        props.put("enable.idempotence", true); // requires acks=all; the client enforces this
        props.put("linger.ms", 50);
        props.put("batch.size", 65536);

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            ProducerRecord<String, String> record = new ProducerRecord<>("orders", "order-42", "hello kafka");
            record.headers().add("source", "demo".getBytes(StandardCharsets.UTF_8));

            RecordMetadata meta = producer.send(record).get();
            System.out.printf("produced to partition=%d offset=%d%n", meta.partition(), meta.offset());
        }
    }
}

import { Kafka } from "kafkajs";

async function main(): Promise<void> {
  const kafka = new Kafka({ brokers: ["localhost:9092"] });
  // kafkajs has no linger.ms; sendBatch (or one messages[] array) is the batching unit.
  // idempotent:true requires maxInFlightRequests <= 5 and acks=-1 (all).
  const producer = kafka.producer({ idempotent: true, maxInFlightRequests: 5 });
  await producer.connect();

  const [meta] = await producer.send({
    topic: "orders",
    acks: -1, // all
    messages: [
      {
        key: "order-42",
        value: "hello kafka",
        headers: { source: "demo" },
      },
    ],
  });
  console.log(`produced to partition=${meta.partition} offset=${meta.baseOffset}`);

  await producer.disconnect();
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});

using System.Text;
using Confluent.Kafka;

var config = new ProducerConfig
{
    BootstrapServers = "localhost:9092",
    Acks = Acks.All,
    EnableIdempotence = true, // requires Acks.All; the client enforces this
    LingerMs = 50,
    BatchSize = 65536,
};

using var producer = new ProducerBuilder<string, string>(config).Build();

var headers = new Headers { { "source", Encoding.UTF8.GetBytes("demo") } };
var result = await producer.ProduceAsync("orders", new Message<string, string>
{
    Key = "order-42",
    Value = "hello kafka",
    Headers = headers,
});
Console.WriteLine($"produced to partition={result.Partition.Value} offset={result.Offset.Value}");

require "rdkafka"

config = Rdkafka::Config.new(
  :"bootstrap.servers"  => "localhost:9092",
  :"acks"               => "all",
  :"enable.idempotence" => true, # requires acks=all; the client enforces this
  :"linger.ms"          => 50,
  :"batch.size"         => 65536,
)
producer = config.producer

handle = producer.produce(
  topic:   "orders",
  payload: "hello kafka",
  key:     "order-42",
  headers: { "source" => "demo" },
)
report = handle.wait # blocks until the broker acks
puts "produced to partition=#{report.partition} offset=#{report.offset}"

use std::time::Duration;

use rdkafka::config::ClientConfig;
use rdkafka::message::{Header, OwnedHeaders};
use rdkafka::producer::{FutureProducer, FutureRecord};
use rdkafka::util::Timeout;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let producer: FutureProducer = ClientConfig::new()
        .set("bootstrap.servers", "localhost:9092")
        .set("acks", "all")
        .set("enable.idempotence", "true") // requires acks=all; the client enforces this
        .set("linger.ms", "50")
        .set("batch.size", "65536")
        .create()?;

    let headers = OwnedHeaders::new().insert(Header { key: "source", value: Some("demo") });
    let record = FutureRecord::to("orders")
        .key("order-42")
        .payload("hello kafka")
        .headers(headers);

    match producer.send(record, Timeout::After(Duration::from_secs(10))).await {
        Ok(delivery) => println!("produced to partition={} offset={}", delivery.partition, delivery.offset),
        Err((err, _)) => return Err(Box::new(err)),
    }
    Ok(())
}