java 118 lines · 3 tabs

Idempotent Kafka Consumer With @KafkaListener and Manual Ack in Spring Boot

Shared by codesnips Aug 2026

3 tabs

package com.shop.payments.messaging;

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;

public record PaymentEvent(
        String eventId,
        String orderId,
        long amountCents,
        String status) {

    @JsonCreator
    public PaymentEvent(
            @JsonProperty("eventId") String eventId,
            @JsonProperty("orderId") String orderId,
            @JsonProperty("amountCents") long amountCents,
            @JsonProperty("status") String status) {
        this.eventId = eventId;
        this.orderId = orderId;
        this.amountCents = amountCents;
        this.status = status;
    }

    public boolean isSuccessful() {
        return "CAPTURED".equalsIgnoreCase(status);
    }
}

package com.shop.payments.messaging;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import com.shop.payments.service.PaymentApplicationService;

@Component
public class PaymentEventListener {

    private static final Logger log = LoggerFactory.getLogger(PaymentEventListener.class);

    private final PaymentApplicationService paymentService;

    public PaymentEventListener(PaymentApplicationService paymentService) {
        this.paymentService = paymentService;
    }

    @KafkaListener(
            topics = "order-payments",
            groupId = "payment-projector",
            containerFactory = "paymentKafkaListenerContainerFactory")
    public void onPaymentEvent(
            @Payload PaymentEvent event,
            @Header(KafkaHeaders.RECEIVED_KEY) String key,
            Acknowledgment ack) {

        log.info("Received payment event id={} order={} key={}", event.eventId(), event.orderId(), key);

        try {
            paymentService.apply(event);
            ack.acknowledge();
        } catch (Exception ex) {
            log.error("Failed to process event {}, will be redelivered", event.eventId(), ex);
            throw ex;
        }
    }
}

package com.shop.payments.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.shop.payments.domain.Order;
import com.shop.payments.domain.OrderRepository;
import com.shop.payments.domain.ProcessedEvent;
import com.shop.payments.domain.ProcessedEventRepository;
import com.shop.payments.messaging.PaymentEvent;

@Service
public class PaymentApplicationService {

    private static final Logger log = LoggerFactory.getLogger(PaymentApplicationService.class);

    private final OrderRepository orderRepository;
    private final ProcessedEventRepository processedEventRepository;

    public PaymentApplicationService(OrderRepository orderRepository,
                                     ProcessedEventRepository processedEventRepository) {
        this.orderRepository = orderRepository;
        this.processedEventRepository = processedEventRepository;
    }

    @Transactional
    public void apply(PaymentEvent event) {
        if (processedEventRepository.existsByEventId(event.eventId())) {
            log.debug("Skipping duplicate event {}", event.eventId());
            return;
        }

        Order order = orderRepository.findById(event.orderId())
                .orElseThrow(() -> new IllegalStateException("Unknown order " + event.orderId()));

        if (event.isSuccessful()) {
            order.markPaid(event.amountCents());
        } else {
            order.markPaymentFailed();
        }

        orderRepository.save(order);
        processedEventRepository.save(new ProcessedEvent(event.eventId()));
    }
}

3 files · java Explain with highlit

This snippet shows how a Spring Boot service consumes an order-payment event from Kafka and updates aggregate state safely, with manual acknowledgment and idempotency baked in. The three tabs move from wire format, to the listener that binds to a topic, to the transactional service that mutates domain state.

In PaymentEvent, the payload is modeled as an immutable record carrying an eventId, orderId, amountCents, and status. The eventId is the key detail: Kafka guarantees at-least-once delivery, so the same message can arrive more than once after a rebalance or a redelivery. Carrying a stable per-event identifier lets the consumer detect and drop duplicates rather than double-applying a payment. Modeling the message as a record keeps deserialization explicit and the object side-effect free.

In PaymentEventListener, the @KafkaListener annotation binds a method to the order-payments topic under a named groupId, which is what makes horizontal scaling work — partitions are distributed across instances sharing that group. The method receives the deserialized PaymentEvent, the raw Kafka key, and an Acknowledgment handle. Because the container is configured for manual ack (AckMode.MANUAL), the offset is only committed via ack.acknowledge() after the service call returns successfully. If processing throws, the offset is not committed and the record is redelivered, so no event is silently lost. The listener deliberately keeps logic thin: it logs, delegates, and acknowledges.

In PaymentApplicationService, the actual state change runs inside @Transactional. It first calls processedEventRepository.existsByEventId(...) to short-circuit duplicates; combined with a unique constraint on event_id, this gives idempotency even under concurrent redelivery, since a racing duplicate insert fails the transaction and rolls back the state change. On a fresh event it loads the Order, applies the payment via markPaid, and records the eventId so future duplicates are rejected. The trade-off is an extra read and write per message, which is cheap relative to the correctness it buys. Developers reach for this pattern whenever a consumer performs non-idempotent mutations — balances, inventory, status transitions — where at-least-once delivery would otherwise corrupt state. Manual ack plus a dedupe table is the pragmatic combination that keeps throughput high while guaranteeing effectively-once processing.

Related snips

package dbutil

import (
  "context"

  "github.com/jackc/pgconn"

Retry Postgres serialization failures with bounded attempts

go postgres transactions

by Leah Thompson 1 tab

graphql

type User {
    id: ID!
    name: String!
    email: String!
    posts: [Post!]!
    createdAt: String!

GraphQL API with Spring Boot

java graphql spring-boot

by David Kumar 3 tabs

ruby

class SignupForm
  include ActiveModel::Model
  include ActiveModel::Attributes

  attribute :account_name, :string
  attribute :email, :string

Shallow Controller, Deep Params: Form Object Pattern

rails activemodel form-object

by codesnips 3 tabs

rust

use crossbeam::channel::unbounded;
use std::thread;

fn main() {
    let (tx, rx) = unbounded();

Crossbeam for advanced concurrent data structures

rust concurrency lock-free

by Marcus Chen 1 tab

javascript

// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
  const success = true;

  setTimeout(() => {
    if (success) {

Promises and async/await patterns for asynchronous JavaScript

javascript promises async-await

by Alex Chang 1 tab

rust

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

Channels (mpsc) for message passing between threads

rust concurrency channels

by Marcus Chen 1 tab

Share this code

Here's the card — post it anywhere.

Idempotent Kafka Consumer With @KafkaListener and Manual Ack in Spring Boot — share card

Link copied