kafka-spring-boot-banking-production-guide

🏦 Production-Grade Spring Boot 3 & Apache Kafka Guide: Banking Payment System

NOTE

Obsidian & Production Ready: This document is an exhaustive, copy-paste-ready guide for building a resilient, production-grade Spring Boot 3.x event-driven banking application with Apache Kafka. It includes full Java snippets, application.yml configs, Mermaid diagrams, failure-mode matrices, and native Obsidian callouts.

---

📌 Table of Contents

1. 1. 🎯 Architectural Flow & Event Lifecycle
2. 2. 🚀 Step-by-Step Project Setup & Configuration
3. 3. 📑 Complete Spring Kafka Annotations Reference
4. 4. 💳 Domain Event Models & DTO Definitions
5. 5. 📤 Production-Grade Transactional & Async Producer
6. 6. 📥 Production-Grade Consumer: Manual Acks & Idempotency
7. 7. 🛡️ Resiliency, Failure Modes & Mitigation Strategies
8. 8. 🧪 Integration Testing with Testcontainers
9. 9. 🧠 Production Troubleshooting & Interview Playbook

---

1. 🎯 Architectural Flow & Event Lifecycle

KEY CONCEPT

In an event-driven banking payment system, financial transactions (PaymentEvent) must achieve zero data loss, strict account-level ordering, idempotent consumption, and non-blocking failure recovery [5-7, 13, 287-289].

flowchart TD
    Client[REST Client / Payment Gateway] -->|Post /api/v1/payments| Producer[Payment Producer Service]
    Producer -->|1. Transact & Send with Key=accountId| Kafka[Kafka Cluster: payment-events Topic]
    
    subgraph Kafka Cluster (5 Partitions)
        P0[Partition 0]
        P1[Partition 1: Account 101 Records]
        P2[Partition 2]
    end
    
    Kafka -->|2. Key Hashing Routes Account 101 to P1| P1
    
    P1 -->|3. Concurrent Consumer Poll| Consumer[Payment Consumer Service]
    
    Consumer -->|4a. Processing Success| DB[(PostgreSQL Ledger)]
    Consumer -->|4b. Transient Error| RetryTopic[payment-events-retry-0 / 1 / 2]
    RetryTopic -->|Retry Exhausted| DLT[payment-events-dlt]
    DLT -->|Alert & Audit| DLTConsumer[@DltHandler / Ops Dashboard]

---

2. 🚀 Step-by-Step Project Setup & Configuration

Step 1: pom.xml Dependencies (Spring Boot 3.x & Spring Kafka 3.x)

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
 
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.2.3</version>
        <relativePath/>
    </parent>
 
    <groupId>com.banking.kafka</groupId>
    <artifactId>payment-processing-service</artifactId>
    <version>1.0.0</version>
    <name>payment-processing-service</name>
    <description>Production-Grade Banking Payment Service with Apache Kafka</description>
 
    <properties>
        <java.version>17</java.version>
    </properties>
 
    <dependencies>
        <!-- Spring Boot Starters -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-validation</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-jpa</artifactId>
        </dependency>
 
        <!-- Spring Kafka -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
 
        <!-- Database -->
        <dependency>
            <groupId>com.h2database</groupId>
            <artifactId>h2</artifactId>
            <scope>runtime</scope>
        </dependency>
 
        <!-- Lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
 
        <!-- Test Dependencies & Testcontainers -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.testcontainers</groupId>
            <artifactId>junit-jupiter</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.testcontainers</groupId>
            <artifactId>kafka</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
 
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

---

Step 2: Production application.yml

server:
  port: 8080
 
spring:
  application:
    name: payment-processing-service
 
  datasource:
    url: jdbc:h2:mem:paymentdb;DB_CLOSE_DELAY=-1
    driver-class-name: org.h2.Driver
    username: sa
    password: 
  jpa:
    hibernate:
      ddl-auto: update
    show-sql: false
 
  kafka:
    bootstrap-servers: localhost:9092
    
    # Producer Production Settings
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      transaction-id-prefix: tx-payment-
      acks: all
      retries: 5
      properties:
        enable.idempotence: true
        max.in.flight.requests.per.connection: 5
        linger.ms: 10
        batch.size: 16384
 
    # Consumer Production Settings
    consumer:
      group-id: payment-execution-group
      auto-offset-reset: earliest
      enable-auto-commit: false # Manual offset management for reliability [114-117]
      isolation-level: read_committed # Ignore aborted transactional events
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        # Poison-Pill Defense with ErrorHandlingDeserializer [216-218]
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
        spring.json.trusted.packages: "com.banking.kafka.*"
        spring.json.use.type.headers: true
        max.poll.records: 500
        max.poll.interval.ms: 300000
 
    listener:
      ack-mode: MANUAL_IMMEDIATE # Commit immediately upon ack.acknowledge() [115, 117]
      concurrency: 3 # 3 parallel consumer threads matching topic partition count [106, 112]
      missing-topics-fatal: false
 
logging:
  level:
    org.springframework.kafka: INFO
    com.banking.kafka: DEBUG

---

3. 📑 Complete Spring Kafka Annotations Reference

NOTE

Below is a comprehensive reference table of all Spring Kafka annotations in execution order, detailing key attributes, production purposes, and critical pitfalls.

Annotation	Target Level	Key Attributes	Production Purpose & Execution Role	Common Pitfalls & Best Practices
@EnableKafka	Class (@Configuration)	N/A	Enables detection of @KafkaListener annotations on Spring beans and sets up the listener container infrastructure [121].	Auto-Configured: Included automatically by Spring Boot auto-configuration. Required if using vanilla Spring [121].
@KafkaListener	Method / Class	topics, groupId, containerFactory, concurrency, autoStartup, properties	Marks a method as an event consumer listener for one or more Kafka topics [120-121].	Group ID: If omitted, defaults to container factory settings [132]. Always ensure concurrency <= partition count [112].
@KafkaHandler	Method	isDefault	Used inside a class-level @KafkaListener to designate target methods for specific payload types [140-141].	Single Default: At most one method can be marked isDefault=true to capture unmapped payload types [141].
@RetryableTopic	Method	attempts, backoff, exclude, topicSuffixingStrategy, dltStrategy	Configures non-blocking retry topics and routes unprocessable events asynchronously to a Dead Letter Topic [287-289, 303-308].	N-1 Rule: attempts = 4 means 1 initial attempt + 3 retries [288-289]. Always set exclude for fatal business errors [308].
@DltHandler	Method	N/A	Marks the method responsible for processing failed records that land on the Dead Letter Topic (DLT) [309, 330].	Failure Strategy: Ensure DLT processing logs, alerts, or stores records without throwing unhandled loops [309, 333].
@SendTo	Method	value	Automatically forwards the return value of a @KafkaListener method to a target output topic [149-150].	Reply Template: Requires container factory to be configured with a valid KafkaTemplate reply template [151-153].
@Payload	Parameter	required	Binds the incoming Kafka message payload body to a method argument [126, 228].	Tombstones: Set required = false when consuming null payloads from log-compacted topics [228].
@Header	Parameter	name, required	Extracts specific Kafka metadata headers (KafkaHeaders.RECEIVED_TOPIC, OFFSET, etc.) [125-126].	Header Names: Use KafkaHeaders constants rather than hardcoded string keys to avoid mismatch [125].
@Transactional	Method / Class	transactionManager, rollbackFor	Executes database and Kafka producer operations within an atomic transaction boundary [187-189].	Dual Managers: Synchronize KafkaTransactionManager with DB transaction manager or use Outbox Pattern [188-191].

---

4. 💳 Domain Event Models & DTO Definitions

package com.banking.kafka.dto;
 
import jakarta.validation.constraints.DecimalMin;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotNull;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
 
import java.math.BigDecimal;
import java.time.Instant;
 
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class PaymentEvent {
 
    @NotBlank(message = "Transaction ID is required")
    private String transactionId;
 
    @NotBlank(message = "Source Account ID is required")
    private String sourceAccountId;
 
    @NotBlank(message = "Target Account ID is required")
    private String targetAccountId;
 
    @NotNull(message = "Amount is required")
    @DecimalMin(value = "0.01", message = "Payment amount must be greater than zero")
    private BigDecimal amount;
 
    @NotBlank(message = "Currency is required")
    private String currency;
 
    private PaymentStatus status;
    private Instant timestamp;
    private String failureReason;
 
    public enum PaymentStatus {
        INITIATED, COMPLETED, FAILED, RETRYING
    }
}

---

5. 📤 Production-Grade Transactional & Async Producer

package com.banking.kafka.producer;
 
import com.banking.kafka.dto.PaymentEvent;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
 
import java.time.Instant;
import java.util.concurrent.CompletableFuture;
 
@Slf4j
@Service
public class PaymentProducer {
 
    private static final String PAYMENT_TOPIC = "payment-events";
    private final KafkaTemplate<String, Object> kafkaTemplate;
 
    public PaymentProducer(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
 
    /**
     * Asynchronous Non-Blocking Event Send with Callback.
     * Uses sourceAccountId as Partition Key to guarantee order per account [13].
     */
    public void sendPaymentEventAsync(PaymentEvent event) {
        event.setTimestamp(Instant.now());
        event.setStatus(PaymentEvent.PaymentStatus.INITIATED);
 
        ProducerRecord<String, Object> record = new ProducerRecord<>(
                PAYMENT_TOPIC,
                event.getSourceAccountId(), // Partition Key for strict ordering [13]
                event
        );
 
        CompletableFuture<SendResult<String, Object>> future = kafkaTemplate.send(record);
 
        future.whenComplete((result, ex) -> {
            if (ex == null) {
                log.info("PaymentEvent [{}] published successfully to Partition {} at Offset {}",
                        event.getTransactionId(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
            } else {
                log.error("Failed to publish PaymentEvent [{}]: {}",
                        event.getTransactionId(), ex.getMessage(), ex);
            }
        });
    }
 
    /**
     * Atomic Transactional Publish.
     * Integrates with spring.kafka.producer.transaction-id-prefix [184, 187].
     */
    @Transactional
    public void sendPaymentEventTransactional(PaymentEvent event) {
        event.setTimestamp(Instant.now());
        log.info("Beginning transactional event dispatch for txId: {}", event.getTransactionId());
 
        kafkaTemplate.send(PAYMENT_TOPIC, event.getSourceAccountId(), event);
        
        // Secondary notification event within the same transaction [187]
        kafkaTemplate.send("payment-audit-log", event.getTransactionId(), 
                "PAYMENT_DISPATCHED_" + event.getAmount());
    }
}

---

6. 📥 Production-Grade Consumer: Manual Acks & Idempotency

JPA Entity for Consumer Idempotency & Deduplication

package com.banking.kafka.entity;
 
import jakarta.persistence.Entity;
import jakarta.persistence.Id;
import jakarta.persistence.Table;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
 
import java.time.Instant;
 
@Entity
@Table(name = "processed_payment_events")
@Data
@NoArgsConstructor
@AllArgsConstructor
public class ProcessedPaymentEvent {
 
    @Id
    private String transactionId;
    private Instant processedAt;
}

Resilient Consumer with Manual Acknowledgment

package com.banking.kafka.consumer;
 
import com.banking.kafka.dto.PaymentEvent;
import com.banking.kafka.entity.ProcessedPaymentEvent;
import com.banking.kafka.repository.ProcessedPaymentRepository;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.DltHandler;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.RetryableTopic;
import org.springframework.kafka.retrytopic.TopicSuffixingStrategy;
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.retry.annotation.Backoff;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
 
import java.time.Instant;
 
@Slf4j
@Service
public class PaymentConsumer {
 
    private final ProcessedPaymentRepository processedPaymentRepository;
 
    public PaymentConsumer(ProcessedPaymentRepository processedPaymentRepository) {
        this.processedPaymentRepository = processedPaymentRepository;
    }
 
    @RetryableTopic(
        attempts = "4", // 1 initial attempt + 3 retries (N-1 order) [288-289]
        backoff = @Backoff(delay = 2000, multiplier = 2.0, maxDelay = 10000), // Exponential backoff [307]
        topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE,
        exclude = { IllegalArgumentException.class, NullPointerException.class } // Fatal exceptions skip retries [308]
    )
    @KafkaListener(topics = "payment-events", groupId = "payment-execution-group")
    @Transactional
    public void processPaymentEvent(@Payload PaymentEvent event,
                                   Acknowledgment acknowledgment,
                                   @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                                   @Header(KafkaHeaders.OFFSET) long offset) {
 
        log.info("Consuming PaymentEvent [TxId: {}] from Partition: {}, Offset: {}",
                event.getTransactionId(), partition, offset);
 
        // 1. Idempotency Check: Prevent duplicate processing across restarts [25, 30]
        if (processedPaymentRepository.existsById(event.getTransactionId())) {
            log.warn("Duplicate PaymentEvent detected [TxId: {}]. Skipping processing.", event.getTransactionId());
            acknowledgment.acknowledge(); // Acknowledge to move offset forward [115, 117]
            return;
        }
 
        // 2. Business Execution Simulation
        if (event.getAmount().doubleValue() > 10000.0) {
            throw new RuntimeException("Transient Core Banking System Timeout for TxId: " + event.getTransactionId());
        }
 
        // 3. Mark Event Processed Idempotently
        processedPaymentRepository.save(new ProcessedPaymentEvent(event.getTransactionId(), Instant.now()));
 
        // 4. Manual Offset Acknowledgment [115, 117]
        acknowledgment.acknowledge();
        log.info("Successfully processed and committed offset for TxId: {}", event.getTransactionId());
    }
 
    @DltHandler
    public void handlePaymentDeadLetter(@Payload PaymentEvent event,
                                        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                                        @Header(KafkaHeaders.OFFSET) long offset) {
        log.error("DEAD LETTER ALERT: PaymentEvent [TxId: {}] routed to DLT [{}] at Offset {}",
                event.getTransactionId(), topic, offset);
        // Alert Operations Team / Persist to Audit DB table for manual intervention
    }
}

---

7. 🛡️ Resiliency, Failure Modes & Mitigation Strategies

Production Failure-Mode Matrix

Scenario / Risk	Root Cause	System Impact	Mitigation Strategy
Transient Outage	DB connection pool exhaustion / Network glitch	Consumer exception thrown during execution	@RetryableTopic offloads message to exponential retry topics (retry-0, 1, 2) [287-289].
Poison Pill Payload	Corrupted JSON / Incompatible POJO class	Deserialization fails before method execution	ErrorHandlingDeserializer catches exception, wraps raw bytes, and routes to DLT without crashing consumer thread [216-218].
Consumer Crash Mid-Batch	Pod killed before offset commit	Last committed offset is behind actual state	Make business operations idempotent (Unique DB constraints / ProcessedEvents table) + use AckMode.MANUAL_IMMEDIATE [25, 115].
Out-of-Order Execution	Partition reassignment / Multi-partition topic	Account balances updated out of chronological order	Set sourceAccountId as Partition Key so all account records hash to the exact same partition [13].
Dual-Write Failure	DB commit succeeds but Kafka producer fails	Data inconsistency between DB and Kafka	Implement Transactional Outbox Pattern or Spring @Transactional synchronization [188-191].

---

8. 🧪 Integration Testing with Testcontainers

package com.banking.kafka;
 
import com.banking.kafka.dto.PaymentEvent;
import com.banking.kafka.producer.PaymentProducer;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.testcontainers.containers.KafkaContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;
 
import java.math.BigDecimal;
import java.util.UUID;
 
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 
@SpringBootTest
@Testcontainers
class PaymentIntegrationTest {
 
    @Container
    static final KafkaContainer kafka = new KafkaContainer(
            DockerImageName.parse("confluentinc/cp-kafka:7.5.0")
    );
 
    @DynamicPropertySource
    static void overrideKafkaProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
    }
 
    @Autowired
    private PaymentProducer paymentProducer;
 
    @Test
    void testPaymentEventProcessingFlow() {
        PaymentEvent event = PaymentEvent.builder()
                .transactionId(UUID.randomUUID().toString())
                .sourceAccountId("ACC-101")
                .targetAccountId("ACC-202")
                .amount(new BigDecimal("250.00"))
                .currency("USD")
                .build();
 
        assertDoesNotThrow(() -> paymentProducer.sendPaymentEventAsync(event));
    }
}

---

9. 🧠 Production Troubleshooting & Interview Playbook

SCENARIO

Q: How do you prevent duplicate processing if a consumer crashes before committing its offset?

Answer: Since Kafka defaults to At-Least-Once delivery, consumer crashes between DB processing and offset commits will re-deliver records upon rebalance [25, 30]. We mitigate this by designing idempotent consumers:

1. Store processed transactionIds in a SQL table with a PRIMARY KEY constraint.
2. Query the repository before processing. If present, invoke ack.acknowledge() immediately and return [115, 117].
3. Use AckMode.MANUAL_IMMEDIATE to minimize the uncommitted offset window [115, 117].

---

SCENARIO

Q: How do non-blocking retries differ from blocking retries in Spring Kafka?

Answer:

• Blocking Retries: The consumer thread sleeps and retries the record on the main topic partition. This blocks subsequent healthy records from being consumed and risks triggering a group rebalance if max.poll.interval.ms is exceeded [161].
• Non-Blocking Retries (@RetryableTopic): Failed records are offloaded to separate internal retry topics (e.g., payment-events-retry-0). The main consumer thread continues processing incoming records on the main partition immediately without blocking [287-289, 303-305].