Kafka idempotencja: gwarancje dostawy w praktyce

# Gwarancje dostarczenia w Kafka: idempotentność producenta i konsumenta w praktyce

Inventory-service przechodzi z przechowywania in-memory na PostgreSQL w celu wsparcia wzorca Outbox Pattern i gwarancji dostarczenia. Dodaj do docker-compose.yaml kontener postgres-inventory z obrazem postgres:17, portami 5432:5432 i zmiennymi POSTGRES_USER=postgres, POSTGRES_PASSWORD=postgres, POSTGRES_DB=inventory_db.

W pom.xml inventory-service dodaj zależności:

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

W application.properties podaj połączenie:

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

Utwórz tabelę 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);

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

Repozytorium z 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);
}

Obsługa KafkaListener w serwisie sprawdza dostępność zapasów i wysyła InventoryReservedEvent lub InventoryRejectedEvent.

Konfiguracja acks i zrozumienie semantyki exactly-once

Parametr spring.kafka.producer.acks zarządza potwierdzeniami: 0 dla at-most-once, 1 dla częściowych, all dla at-least-once z replikacją na wszystkich followerach.

Exactly-once delivery w systemach rozproszonych jest niemożliwa ze względu na Problem dwóch generałów. Realizuje się jako exactly-once semantics: at-least-once delivery + idempotentność producenta i konsumenta.

Idempotentność producenta: ProducerId, Epoch i sequence number

Idempotentność zapobiega duplikatom przy ponownych próbach z powodu utraty ack. Włącza się ją linią:

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

Broker przypisuje połączeniu parę (ProducerId, ProducerEpoch):

• ProducerId: unikalny identyfikator producenta
• ProducerEpoch: inkrement przy restarcie, ochrona przed zombie-producentami

Każda wiadomość jest numerowana sequence number. Broker odrzuca duplikaty w ramach ProducerId/Epoch/sequence.

Scenariusz zombie-producenta:

• Producent się wywala, ale „ożywa” po restarcie nowego
• Stary kontynuuje z wysokim sequence number
• Nowy zaczyna od 1 — ProducerEpoch ich rozróżnia

Transactional.id zapewnia zachowanie ProducerId przy restarcie (Spring wywołuje initTransactions()). Bez niego — nowy losowy ProducerId.

Krytyczne parametry idempotentności

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

Naruszenie wyłącza idempotentność. Zastosuj w order-service i inventory-service.

Idempotentność konsumenta przeciw podwójnemu przetwarzaniu

Konsument może przetworzyć wiadomość dwukrotnie przy awarii po commit offset, ale przed zakończeniem logiki biznesowej. Dla inventory-service:

• Odczytuje OrderPlacedEvent
• deductStock() w PostgreSQL (atomowo)
• Wysyła InventoryReservedEvent
• Awaria
• Ponownie odczytuje offset, powtarza deductStock() — ale quantity już zmniejszone, UPDATE nie zadziała

To zapewnia idempotentność na poziomie logiki biznesowej poprzez warunkowy UPDATE.

Zalety podejścia:

• Brak dodatkowego przechowywania offsetów
• Wykorzystuje transakcje ACID bazy danych
• Prosta implementacja poprzez WHERE quantity >= :quantity

Co ważne

• Idempotentność producenta: enable.idempotence=true + acks=all + retries>0 + max.in.flight≤5
• ProducerEpoch: ochrona przed zombie-producentami przy restarcie
• Transactional.id: zachowanie ProducerId, wymaga initTransactions()
• Konsument: logika biznesowa z warunkowymi UPDATE w bazie danych zapewnia idempotentność
• Exactly-once semantics: at-least-once + idempotentność, nie dosłowne exactly-once

— Editorial Team