Kafka Idempotenz: Zustellgarantien in der Praxis

Kafka-Zustellgarantien: Producer- und Consumer-Idempotenz in der Praxis

Der inventory-service wechselt vom In-Memory-Speicher zu PostgreSQL, um das Outbox-Pattern und Zustellgarantien zu unterstützen. Fügen Sie einen postgres-inventory-Container zu docker-compose.yaml hinzu mit image postgres:17, ports 5432:5432 und Umgebungsvariablen POSTGRES_USER=postgres, POSTGRES_PASSWORD=postgres, POSTGRES_DB=inventory_db.

Im pom.xml des inventory-service fügen Sie die folgenden Abhängigkeiten hinzu:

<dependency>
  <groupId>org.postgresql</groupId>
  <artifactId>postgresql</artifactId>
  <scope>runtime</scope>
</dependency>
<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>

In application.properties legen Sie die Verbindung fest:

spring.datasource.url=jdbc:postgresql://localhost:5432/inventory_db
spring.datasource.username=postgres
spring.datasource.password=postgres

Erstellen Sie die inventory-Tabelle:

CREATE TABLE inventory (
    id BIGSERIAL PRIMARY KEY,
    product_name VARCHAR(256) NOT NULL UNIQUE,
    quantity INT NOT NULL CHECK (quantity >= 0)
);

INSERT INTO inventory (product_name, quantity) VALUES
('Smartphone', 5),
('Tablet', 10),
('Desktop', 6);

Inventory-Entity:

@Entity
@Table(name = "inventory")
public class Inventory {
    @Id
    @Column(name = "id")
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    @Column(name = "product_name", nullable = false, unique = true, length = 256)
    private String productName;

    @Column(name = "quantity", nullable = false)
    private Integer quantity;

    public Inventory(String productName, Integer quantity) {
        this.productName = productName;
        this.quantity = quantity;
    }
    // getters, setters
}

Repository mit deductStock:

public interface InventoryRepository extends JpaRepository<Inventory, Long> {
    @Modifying
    @Query("UPDATE Inventory i SET i.quantity = i.quantity - :quantity WHERE i.productName = :productName AND i.quantity >= :quantity")
    int deductStock(@Param("productName") String productName, @Param("quantity") Integer quantity);
}

Der KafkaListener-Handler im Service prüft die Verfügbarkeit des Lagerbestands und sendet InventoryReservedEvent oder InventoryRejectedEvent.

Einrichtung von acks und Verständnis der exactly-once-Semantik

Der Parameter spring.kafka.producer.acks steuert die Bestätigungen: 0 für at-most-once, 1 für teilweise, all für at-least-once mit Replikation auf allen Followern.

Exactly-once-Zustellung in verteilten Systemen ist aufgrund des Zwei-Generäle-Problems unmöglich. Sie wird als exactly-once-Semantik umgesetzt: at-least-once-Zustellung + Producer- und Consumer-Idempotenz.

Producer-Idempotenz: ProducerId, Epoch und Sequenznummer

Idempotenz verhindert Duplikate bei Wiederholungen durch verlorene acks. Sie wird aktiviert durch:

spring.kafka.producer.properties.enable.idempotence=true

Der Broker weist der Verbindung ein (ProducerId, ProducerEpoch)-Paar zu:

• ProducerId: eindeutige Producer-Kennung
• ProducerEpoch: inkrementiert sich bei Neustart, Schutz vor Zombie-Producern

Jede Nachricht erhält eine Sequenznummer. Der Broker lehnt Duplikate innerhalb von ProducerId/Epoch/Sequenz ab.

Zombie-Producer-Szenario:

• Producer stürzt ab, "wird aber" nach Neustart eines neuen wiederbelebt
• Der alte fährt mit hoher Sequenznummer fort
• Der neue beginnt bei 1 – ProducerEpoch unterscheidet sie

Transactional.id sorgt für die Erhaltung der ProducerId bei Neustart (Spring ruft initTransactions() auf). Ohne sie – neue zufällige ProducerId.

Kritische Idempotenz-Parameter

spring.kafka.producer.acks=all
spring.kafka.producer.retries=2147483647  # >0
spring.kafka.producer.properties.max.in.flight.requests.per.connection=5  # ≤5
spring.kafka.producer.properties.enable.idempotence=true

Verstöße deaktivieren die Idempotenz. Wenden Sie sie auf order-service und inventory-service an.

Consumer-Idempotenz gegen doppelte Verarbeitung

Der Consumer kann eine Nachricht doppelt verarbeiten, wenn er nach dem Commit des Offsets abstürzt, aber vor Abschluss der Geschäftslogik. Für inventory-service:

• Liest OrderPlacedEvent
• deductStock() in PostgreSQL (atomar)
• Sendet InventoryReservedEvent
• Absturz
• Liest Offset erneut, wiederholt deductStock() – aber quantity bereits reduziert, UPDATE greift nicht

Das gewährleistet Idempotenz auf Geschäftslogik-Ebene durch bedingtes UPDATE.

Vorteile des Ansatzes:

• Kein zusätzlicher Offset-Speicher
• Nutzt DB-ACID-Transaktionen
• Einfache Umsetzung über WHERE quantity >= :quantity

Wichtige Punkte

• Producer-Idempotenz: enable.idempotence=true + acks=all + retries>0 + max.in.flight≤5
• ProducerEpoch: Schutz vor Zombie-Producern bei Neustart
• Transactional.id: erhält ProducerId, erfordert initTransactions()
• Consumer: Geschäftslogik mit bedingtem DB-UPDATE gewährleistet Idempotenz
• Exactly-once-Semantik: at-least-once + Idempotenz, keine wörtliche exactly-once

— Editorial Team