Kafka Producer Exactly-Once Semantics in Spring Boot: Idempotent Producer, Transactions & Offset Management (2026)

Kafka's exactly-once semantics (EOS) requires three cooperative pieces: the idempotent producer (enable.idempotence=true) to deduplicate retried batches at the broker, producer transactions to atomically publish across partitions, and read_committed isolation on consumers. In Spring Boot, wire these up with KafkaTransactionManager and @Transactional. For dual-write safety between your database and Kafka, use the Transactional Outbox pattern. Monitor abort rates and consumer lag religiously in production.

Table of Contents

1. Why Exactly-Once Matters
2. Kafka Delivery Semantics Deep Dive
3. Idempotent Producer (enable.idempotence=true)
4. Producer Transactions in Kafka
5. Spring Boot KafkaTransactionManager Configuration
6. Transactional Outbox Pattern with Kafka
7. Consumer Side: Read-Committed Isolation Level
8. Offset Management Strategies
9. Monitoring Transactions
10. Common Pitfalls and Anti-Patterns
11. Production Checklist

spring:
  kafka:
    producer:
      bootstrap-servers: localhost:9092
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      properties:
        enable.idempotence: true
        max.in.flight.requests.per.connection: 5
        acks: all
        retries: 2147483647
        delivery.timeout.ms: 120000

@Configuration
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        config.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx-producer-");
        config.put(ProducerConfig.ACKS_CONFIG, "all");
        config.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
        config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
        DefaultKafkaProducerFactory<String, Object> factory =
            new DefaultKafkaProducerFactory<>(config);
        factory.setTransactionIdPrefix("order-tx-");
        return factory;
    }

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    @Bean
    public KafkaTransactionManager<String, Object> kafkaTransactionManager() {
        return new KafkaTransactionManager<>(producerFactory());
    }
}

@Service
@Slf4j
public class OrderService {

    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final OrderRepository orderRepository;

    @Transactional("kafkaTransactionManager")
    public void processOrder(Order order) {
        // Save to DB and publish event atomically
        orderRepository.save(order);

        OrderCreatedEvent event = OrderCreatedEvent.builder()
            .orderId(order.getId())
            .customerId(order.getCustomerId())
            .totalAmount(order.getTotalAmount())
            .timestamp(Instant.now())
            .build();

        kafkaTemplate.send("orders.created", order.getId().toString(), event);
        kafkaTemplate.send("notifications.email", order.getCustomerId(), event);

        log.info("Order {} published transactionally", order.getId());
    }
}

spring:
  kafka:
    producer:
      transaction-id-prefix: order-tx-
      properties:
        transaction.timeout.ms: 60000      # broker-side transaction timeout
        max.block.ms: 10000                # max wait for send() metadata
    listener:
      ack-mode: manual_immediate
      concurrency: 3

@Entity
@Table(name = "outbox_events")
@Data
@Builder
public class OutboxEvent {
    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private String id;

    @Column(nullable = false)
    private String aggregateType;

    @Column(nullable = false)
    private String aggregateId;

    @Column(nullable = false)
    private String eventType;

    @Column(columnDefinition = "TEXT", nullable = false)
    private String payload;

    @Column(nullable = false)
    @Enumerated(EnumType.STRING)
    private OutboxStatus status;  // PENDING, PUBLISHED, FAILED

    @Column(nullable = false)
    private Instant createdAt;

    private Instant publishedAt;
    private int retryCount;
}

@Component
@Slf4j
public class OutboxPublisher {

    private final OutboxEventRepository outboxRepo;
    private final KafkaTemplate<String, Object> kafkaTemplate;

    @Scheduled(fixedDelay = 1000)
    @Transactional
    public void publishPendingEvents() {
        List<OutboxEvent> events = outboxRepo
            .findTop100ByStatusOrderByCreatedAtAsc(OutboxStatus.PENDING);

        events.forEach(event -> {
            try {
                kafkaTemplate.send(
                    resolveTopicName(event.getAggregateType()),
                    event.getAggregateId(),
                    event.getPayload()
                );
                event.setStatus(OutboxStatus.PUBLISHED);
                event.setPublishedAt(Instant.now());
            } catch (Exception e) {
                event.setRetryCount(event.getRetryCount() + 1);
                if (event.getRetryCount() >= 5) {
                    event.setStatus(OutboxStatus.FAILED);
                }
                log.error("Failed to publish outbox event {}", event.getId(), e);
            }
        });
        outboxRepo.saveAll(events);
    }
}

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "order-consumer-group");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        config.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        config.put(JsonDeserializer.TRUSTED_PACKAGES, "com.example.events");
        return new DefaultKafkaConsumerFactory<>(config);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }
}

@Component
@Slf4j
public class OrderConsumer {

    @KafkaListener(
        topics = "orders.created",
        groupId = "order-processor",
        containerFactory = "kafkaListenerContainerFactory"
    )
    public void consumeOrder(
            ConsumerRecord<String, OrderCreatedEvent> record,
            Acknowledgment ack) {
        try {
            OrderCreatedEvent event = record.value();
            log.info("Processing order {} from partition {} offset {}",
                event.getOrderId(), record.partition(), record.offset());

            processOrderEvent(event);
            ack.acknowledge();  // Commit offset only on success

        } catch (NonRetryableException e) {
            log.error("Non-retryable error, sending to DLQ", e);
            ack.acknowledge();  // Acknowledge to skip poison pill
        } catch (Exception e) {
            log.error("Retryable error processing order", e);
            // Do NOT acknowledge - message will be redelivered
            throw e;
        }
    }
}

@Configuration
public class KafkaMetricsConfig {

    @Bean
    public MeterRegistryCustomizer<MeterRegistry> kafkaMetrics(
            KafkaTemplate<String, Object> kafkaTemplate) {
        return registry -> {
            // Transaction success/failure counters
            Counter.builder("kafka.transactions.committed")
                .description("Number of committed Kafka transactions")
                .register(registry);

            Counter.builder("kafka.transactions.aborted")
                .description("Number of aborted Kafka transactions")
                .register(registry);

            // Producer retry gauge
            Gauge.builder("kafka.producer.record.retry.total",
                    kafkaTemplate, t -> getProducerMetric(t, "record-retry-total"))
                .description("Total producer record retries")
                .register(registry);
        };
    }
}

management:
  endpoints:
    web:
      exposure:
        include: health, metrics, prometheus
  metrics:
    tags:
      application: ${spring.application.name}
    export:
      prometheus:
        enabled: true
  endpoint:
    health:
      show-details: always

// BAD - transaction commits even though processing failed
@Transactional("kafkaTransactionManager")
public void badMethod(Order order) {
    try {
        kafkaTemplate.send("orders.created", order.getId().toString(), order);
        riskyBusinessLogic(order);  // throws RuntimeException
    } catch (Exception e) {
        log.error("Something went wrong", e);
        // Exception is swallowed - Kafka transaction WILL commit!
    }
}

// GOOD - let the exception propagate to trigger rollback
@Transactional("kafkaTransactionManager")
public void goodMethod(Order order) {
    kafkaTemplate.send("orders.created", order.getId().toString(), order);
    riskyBusinessLogic(order);  // exception propagates, transaction aborted
}

@SpringBootTest
@Testcontainers
class OrderServiceEosIntegrationTest {

    @Container
    static KafkaContainer kafka = new KafkaContainer(
        DockerImageName.parse("confluentinc/cp-kafka:7.6.0"))
        .withKraft();

    @DynamicPropertySource
    static void kafkaProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
    }

    @Autowired OrderService orderService;

    @Test
    void shouldPublishExactlyOnceOnRetry() throws Exception {
        Order order = Order.builder()
            .id(UUID.randomUUID())
            .customerId("cust-42")
            .totalAmount(new BigDecimal("99.99"))
            .build();

        orderService.processOrder(order);

        // Consume from orders.created and assert exactly one record
        ConsumerRecords<String, ?> records = pollKafka("orders.created", 1);
        assertThat(records.count()).isEqualTo(1);
    }
}

Leave a Comment

Related Posts