Kafka-Producer: acks, Idempotenz und Batching

1. Übersicht

Eine Nachricht zu Kafka zu senden, benötigt nur eine Zeile send(). Die schwierige Arbeit liegt in den begleitenden Fragen: Wurde die Nachricht wirklich persistiert? Wird sie bei einem Producer-Retry doppelt geschrieben? Und wie versendest du Zehntausende Nachrichten pro Sekunde, ohne den Broker zu überlasten?

Dieser Artikel behandelt drei Konfigurationsgruppen, die diese drei Fragen beantworten: acks, enable.idempotence und Batching (batch.size, linger.ms, compression.type). Nach dem Lesen wirst du wissen, wie du einen Producer sowohl sicher als auch schnell konfigurierst – mit reinem Kafka-Client oder Spring Boot 3.x – und wie du ihn testest.

2. Einrichtung

Mit reinem Java brauchst du nur kafka-clients:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.9.0</version>
</dependency>

Mit Spring Boot 3.x verwendest du spring-kafka. Die Versionsverwaltung erfolgt durch Spring Boot, und es zieht automatisch kafka-clients mit sich:

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

Durchgehend in diesem Artikel versenden wir ein Bestellereignis:

public record Order(String id, String customerId, BigDecimal total) {

    public String toJson() {
        return """
            {"id":"%s","customerId":"%s","total":%s}""".formatted(id, customerId, total);
    }
}

3. acks: Wann gilt eine Nachricht als versendet?

acks legt fest, wie viele Bestätigungen der Producer vom Broker warten muss, bevor er einen Request als erfolgreich einstuft.

Wert	Bedeutung	Risiko
acks=0	Wartet überhaupt nicht	Nachrichtenverlust ohne Benachrichtigung
acks=1	Wartet, bis der Leader geschrieben hat	Nachrichtenverlust, wenn der Leader abstürzt, bevor der Follower die Nachricht repliziert
acks=all (-1)	Wartet, bis alle Replicas in der ISR geschrieben haben	Etwas höhere Latenz

Beispiel für die Konfiguration mit reinem Client:

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.ACKS_CONFIG, "all");

try (var producer = new KafkaProducer<String, String>(props)) {
    var order = new Order("o-1001", "c-42", new BigDecimal("250000"));
    RecordMetadata meta = producer
        .send(new ProducerRecord<>("orders", order.customerId(), order.toJson()))
        .get();
    System.out.printf("partition=%d offset=%d%n", meta.partition(), meta.offset());
}

Der Aufruf von .get() macht das Versenden synchron. Hier tun wir das nur zur Ausgabe des Ergebnisses. Mit acks=all gibt .get() erst zurück, nachdem alle Replicas in der ISR geschrieben haben:

partition=2 offset=17

Wir verwenden customerId als Schlüssel, daher gehen alle Bestellungen desselben Kunden in dieselbe Partition und behalten die richtige Reihenfolge.

3.1. acks=all ist nicht ausreichend ohne min.insync.replicas

acks=all bedeutet „alle Replicas, die gerade in der ISR sind”. Angenommen, die ISR schrumpft auf nur den Leader. Dann unterscheidet sich acks=all nicht von acks=1. Du musst zusätzlich im Topic oder Broker konfigurieren:

kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders \
  --partitions 6 --replication-factor 3 \
  --config min.insync.replicas=2

Mit replication-factor=3 und min.insync.replicas=2 müssen mindestens 2 Replikationen beim Schreiben vorhanden sein. Falls die ISR unter 2 fällt, erhält der Producer die Exception NotEnoughReplicasException, anstatt heimlich nur eine Kopie zu schreiben.

Hinweis: Setze min.insync.replicas nicht gleich dem replication-factor. Dann reicht ein Broker-Neustart aus, um das Topic für alle Schreibvorgänge abzulehnen.

4. Idempotenz: Retry ohne Duplikate

acks=all bedeutet Retry. Betrachte dieses Szenario: Der Broker hat die Nachricht geschrieben, aber die Antwort geht auf dem Rückweg verloren. Der Producer denkt, der Versand ist fehlgeschlagen, und versendet die Nachricht erneut – sie wird zweimal geschrieben.

Idempotente Producer lösen genau dieses Szenario. Wenn du enable.idempotence=true aktivierst:

• Der Broker vergibt eine Producer-ID (PID) an den Producer.
• Jeder Batch, der an eine Partition gesendet wird, trägt eine aufsteigende Sequenznummer.
• Der Broker verwirft Batches mit bereits gesehenen Sequenznummern und lehnt Batches ab, die die Sequenznummer überspringen. Dadurch wird die Reihenfolge beibehalten, auch wenn mehrere Requests gleichzeitig unterwegs sind.

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);

Die Bedeutung jeder Zeile:

• ENABLE_IDEMPOTENCE_CONFIG aktiviert den Mechanismus mit PID und Sequenznummer.
• Idempotenz erzwingt acks=all. Falls du acks=1 zusammen mit enable.idempotence=true explizit setzt, wirft der Producer sofort beim Erstellen eine ConfigException.
• max.in.flight.requests.per.connection muss ≤ 5 sein. Innerhalb dieses Limits garantiert Kafka die Reihenfolge innerhalb der Partition.
• retries wird sehr groß gesetzt. Die tatsächliche Retry-Zeit wird durch delivery.timeout.ms begrenzt (Standardwert 2 Minuten).

Seit Kafka 3.0 sind acks=all und enable.idempotence=true die Standardwerte. Dennoch solltest du sie explizit deklarieren. Wenn nicht, reicht aus, dass jemand acks=1 setzt, und die Idempotenz wird stillschweigend deaktiviert.

4.1. Was Idempotenz nicht kann

Dies ist ein häufiges Missverständnis:

• Sie verhindert Duplikate nur innerhalb einer Producer-Session, pro Partition. Wenn die Anwendung neu startet, erhält der Producer eine neue PID. Nachrichten, die nach einem Neustart erneut gesendet werden, werden immer noch doppelt geschrieben.
• Sie verhindert keine Duplikate, die durch deinen Code entstehen, z. B. wenn du send() zweimal aufrufst, etwa bei eigenem Retry im catch.
• Sie ermöglicht keine atomaren Schreibvorgänge auf mehrere Partitionen.

Diese Fälle erfordern Transaktionen (transactional.id) oder einen Consumer, der Duplikate basierend auf orderId entfernen kann.

5. Batching: Latenz gegen Durchsatz eintauschen

Der Producer versendet nicht jede Nachricht einzeln. send() platziert die Nachricht nur in einem In-Memory-Buffer, organisiert nach Partition. Ein Background-Thread (Sender) fasst Datensätze zu Batches zusammen und versendet sie. Drei Hauptparameter:

Parameter	Standard	Aufgabe
batch.size	16384 Byte	Maximale Größe eines Batch pro Partition
linger.ms	0 (5 ab Kafka 4.0)	Wartezeit zum Füllen eines Batch
compression.type	none	Komprimierung des gesamten Batch: gzip, snappy, lz4, zstd
buffer.memory	32 MB	Speicher für alle ausstehenden Batches

Ein Batch wird versendet, wenn er mit batch.size voll ist oder wenn linger.ms verstrichen ist – je nachdem, was zuerst eintritt. Mit linger.ms=0 werden Batches meist sofort versendet, enthalten also meist nur wenige Datensätze.

props.put(ProducerConfig.BATCH_SIZE_CONFIG, 64 * 1024);
props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

Diese Konfiguration akzeptiert bis zu 10 ms zusätzlicher Latenz. Im Gegenzug trägt jeder Request bis zu 64 KB komprimierte Daten. Bei wiederholten JSON-Daten wie Bestellereignissen reduzieren lz4 oder zstd oft deutlich die Netzwerkbandbreite und den Speicher auf dem Broker. Batch-Kompression ist viel effektiver als einzelne Nachrichtenkompression, weil größere Batches mehr wiederholte Muster bieten.

Du kannst dies anhand der Producer-Metriken überprüfen:

producer.metrics().forEach((name, metric) -> {
    if (name.name().equals("batch-size-avg") || name.name().equals("records-per-request-avg")) {
        System.out.println(name.name() + " = " + metric.metricValue());
    }
});

Versende 10.000 Bestellungen hintereinander mit linger.ms=0, dann wiederhole es mit linger.ms=10. Du wirst sehen, dass records-per-request-avg deutlich ansteigt:

# linger.ms=0
batch-size-avg = 1830.4
records-per-request-avg = 21.7
# linger.ms=10
batch-size-avg = 41022.9
records-per-request-avg = 486.2

Die exakten Zahlen hängen von der Maschine und der Last ab, aber der Trend ist immer gleich.

Hinweis: Wenn der Buffer voll wird (langsamer Broker oder Netzwerkstau), blockiert send() bis zu max.block.ms (Standard 60 Sekunden) und wirft dann TimeoutException. Rufe send() nicht im HTTP-Request-Handler auf, ohne diese Möglichkeit zu berücksichtigen.

Hinweis: delivery.timeout.ms muss ≥ linger.ms + request.timeout.ms sein. Andernfalls lehnt der Producer die Initialisierung ab.

6. Konfiguration in Spring Boot 3.x

Spring Boot hat eigene Properties für gängige Parameter. Übrige Parameter gehen über properties:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      acks: all
      batch-size: 64KB
      compression-type: lz4
      buffer-memory: 64MB
      properties:
        enable.idempotence: true
        linger.ms: 10
        max.in.flight.requests.per.connection: 5
        delivery.timeout.ms: 120000

Der Publisher verwendet KafkaTemplate. Seit Spring Kafka 3.0 gibt send() ein CompletableFuture zurück:

@Service
public class OrderEventPublisher {

    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);
    private final KafkaTemplate<String, String> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public CompletableFuture<SendResult<String, String>> publish(Order order) {
        return kafkaTemplate.send("orders", order.customerId(), order.toJson())
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("Versand der Bestellung {} fehlgeschlagen", order.id(), ex);
                } else {
                    var meta = result.getRecordMetadata();
                    log.info("Bestellung {} -> Partition {} Offset {}",
                        order.id(), meta.partition(), meta.offset());
                }
            });
    }
}

Beachte drei Punkte:

• Der Callback läuft auf dem Sender-Thread des Producers. Führe im whenComplete keine schwere Arbeit aus und rufe keine blockierenden Operationen auf, da dies alle anderen Versendungen verlangsamt.
• Wenn der Callback einen Fehler erhält, hat der Producer bereits alle Retries innerhalb des delivery.timeout.ms aufgebraucht. Versuche nicht, von Hand send() erneut aufzurufen. Das ist genau der Grund für Duplikate, die Idempotenz nicht verhindern kann.
• Das Zurückgeben eines CompletableFuture lässt den Aufrufer entscheiden, ob er warten möchte oder nicht.

7. Test des Producer-Verhaltens

kafka-clients bietet MockProducer zum Testen ohne Broker. Um ihn zu nutzen, trennst du die Logik in eine Klasse, die eine Producer-Schnittstelle akzeptiert:

public class OrderSender {

    private final Producer<String, String> producer;

    public OrderSender(Producer<String, String> producer) {
        this.producer = producer;
    }

    public Future<RecordMetadata> send(Order order) {
        return producer.send(new ProducerRecord<>("orders", order.customerId(), order.toJson()));
    }

    public static Properties reliableConfig(String bootstrap) {
        Properties p = new Properties();
        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
        p.put(ProducerConfig.ACKS_CONFIG, "all");
        p.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        p.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
        p.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        p.put(ProducerConfig.BATCH_SIZE_CONFIG, 64 * 1024);
        p.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
        return p;
    }
}

Test mit JUnit 5 und AssertJ:

class OrderSenderTest {

    @Test
    void whenSendingOrders_thenKeyIsCustomerId() {
        var mock = new MockProducer<>(true, new StringSerializer(), new StringSerializer());
        var sender = new OrderSender(mock);

        sender.send(new Order("o-1", "c-42", new BigDecimal("100")));
        sender.send(new Order("o-2", "c-42", new BigDecimal("200")));

        assertThat(mock.history())
            .hasSize(2)
            .extracting(ProducerRecord::key)
            .containsOnly("c-42");
    }

    @Test
    void whenBrokerFails_thenFutureCompletesExceptionally() {
        var mock = new MockProducer<>(false, new StringSerializer(), new StringSerializer());
        var sender = new OrderSender(mock);

        Future<RecordMetadata> future = sender.send(new Order("o-3", "c-7", BigDecimal.TEN));
        mock.errorNext(new NotEnoughReplicasException("ISR < min.insync.replicas"));

        assertThatThrownBy(future::get)
            .hasCauseInstanceOf(NotEnoughReplicasException.class);
    }

    @Test
    void reliableConfig_isAccepted() {
        var props = OrderSender.reliableConfig("localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        var config = new ProducerConfig(props);

        assertThat(config.getString(ProducerConfig.ACKS_CONFIG)).isEqualTo("-1");
        assertThat(config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG)).isTrue();
    }
}

Jeder Test überprüft ein Verhalten:

• Der erste Test nutzt autoComplete=true, daher gelingt jeder send() sofort. history() ermöglicht die Überprüfung des Schlüssels und bestätigt, dass die Reihenfolge pro Kunde gewahrt bleibt.
• Der zweite Test nutzt autoComplete=false und setzt dann errorNext(...), um das Broker-Ablehnen von Schreibvorgängen zu simulieren. So testest du, wie dein Code mit Fehlern umgeht.
• Der letzte Test initialisiert ProducerConfig, damit Kafka die Konfigurationskombination selbst validiert. Eine ungültige Kombination, etwa acks=1 mit Idempotenz, führt zu ConfigException bei der Erstellung, nicht beim Deployment. Kafka normalisiert all zu "-1", daher wird auf "-1" verglichen.

8. Wann solltest du lockerer werden?

Eine „maximal sichere” Konfiguration ist nicht immer korrekt:

• Metriken, Logs, Clickstreams: Wenige verlorene Einträge sind akzeptabel. Nutze acks=1, hohes linger.ms und Kompression mit zstd für den besten Durchsatz.
• Systeme, die geringe pro-Nachricht-Latenz benötigen (z. B. Echtzeitquotierungen): Halte linger.ms klein (0–2 ms). Erwarte dann aber nicht, dass Batches groß werden.
• Zahlungen, Bestellungen: Nutze immer acks=all, Idempotenz und min.insync.replicas=2. Wenn du Lesen–Verarbeiten–Schreiben genau einmal brauchst, erwäge Transaktionen.

Hinweis: max.in.flight.requests.per.connection=1 „zum Sicherheitsgebot” zu setzen, ist unnötig, wenn Idempotenz aktiviert ist. Es senkt nur den Durchsatz.

9. Fazit

Drei Konfigurationsgruppen lösen drei verschiedene Probleme:

• acks=all kombiniert mit min.insync.replicas stellt keinen Nachrichtenverlust sicher.
• enable.idempotence stellt keine Duplikate beim Retry sicher und bewahrt die Reihenfolge innerhalb der Partition. Sein Geltungsbereich ist nur eine Producer-Session.
• batch.size, linger.ms und compression.type entscheiden über den Durchsatz – auf Kosten etwas Latenz.

Deklariere diese Werte explizit, statt dich auf Standardwerte zu verlassen. Validiere sie durch Tests, um Überraschungen beim Deployment zu vermeiden. Und mache dein eigenes Retry für send() nicht, wenn der Producer das bereits für dich tut.