Kafka idempotentnost: záruky doručení v praxi

# Garance doručení v Kafka: Idempotence producentu a konzumenta v praxi

Inventory-service přechází z úložiště v paměti na PostgreSQL pro podporu Outbox Patternu a garancí doručení. Přidejte do docker-compose.yaml kontejner postgres-inventory s image postgres:17, porty 5432:5432 a proměnnými POSTGRES_USER=postgres, POSTGRES_PASSWORD=postgres, POSTGRES_DB=inventory_db.

V pom.xml inventory-service přidejte závislosti:

<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>

V application.properties uveďte připojení:

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

Vytvořte tabulku inventory:

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);

Entita Inventory:

@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
}

Repozitář s 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);
}

Obsluha KafkaListener v servisu kontroluje dostupnost zásob a odesílá InventoryReservedEvent nebo InventoryRejectedEvent.

Nastavení acks a porozumění exactly-once semantics

Parametr spring.kafka.producer.acks řídí potvrzení: 0 pro at-most-once, 1 pro částečné, all pro at-least-once s replikací na všech followerech.

Exactly-once doručení v distribuovaných systémech není možné kvůli problému dvou generálů. Realizuje se jako exactly-once semantics: at-least-once doručení + idempotence producentu a konzumenta.

Idempotence producentu: ProducerId, Epoch a sequence number

Idempotence zabraňuje duplikátům při opakování kvůli ztrátě ack. Zapne se řádkem:

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

Broker přiřadí spojení páru (ProducerId, ProducerEpoch):

• ProducerId: unikátní identifikátor producentu
• ProducerEpoch: inkrement při restartu, ochrana před zombie-produkenty

Každá zpráva má sequence number. Broker odmítne duplikáty v rámci ProducerId/Epoch/sequence.

Scénář zombie-producenta:

• Producent havaruje, ale „ožije“ po restartu nového
• Starý pokračuje s vysokým sequence number
• Nový začíná od 1 — ProducerEpoch je odlišný

Transactional.id zajišťuje zachování ProducerId při restartu (Spring volá initTransactions()). Bez něj — nový náhodný ProducerId.

Kritické parametry idempotence

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

Porušení vypne idempotenci. Aplikujte na order-service a inventory-service.

Idempotence konzumenta proti dvojitému zpracování

Konzument může zpracovat zprávu dvakrát při havárii po commitu offsetu, ale před dokončením obchodní logiky. Pro inventory-service:

• Čte OrderPlacedEvent
• deductStock() v PostgreSQL (atomárně)
• Odesílá InventoryReservedEvent
• Havárie
• Znovu čte offset, opakuje deductStock() — ale quantity už sníženo, UPDATE neproběhne

To zajišťuje idempotenci na úrovni obchodní logiky prostřednictvím podmíněného UPDATE.

Výhody přístupu:

• Žádné dodatečné ukládání offsetů
• Využívá ACID transakce databáze
• Jednoduchá implementace přes WHERE quantity >= :quantity

Co je důležité

• Idempotence producentu: enable.idempotence=true + acks=all + retries>0 + max.in.flight≤5
• ProducerEpoch: ochrana před zombie-produkenty při restartu
• Transactional.id: zachování ProducerId, vyžaduje initTransactions()
• Konzument: obchodní logika s podmíněnými UPDATE v databázi zajišťuje idempotenci
• Exactly-once semantics: at-least-once + idempotence, ne striktní exactly-once

— Editorial Team