<dependencies>
    <!-- Spring Kafka -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
 
    <!-- Jackson for JSON serialization -->
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>
</dependencies>

# application.yml
spring:
  kafka:
    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
    # Producer and consumer configs are defined programmatically below,
    # but bootstrap-servers is shared and can be injected via @Value.

package com.example.kafka.config;
 
import com.example.kafka.dto.OrderEvent;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;
 
import java.util.HashMap;
import java.util.Map;
 
/**
 * Production-ready Kafka Producer configuration.
 *
 * Design decisions:
 * - StringSerializer for keys: deterministic partition routing via murmur2 hash.
 * - JsonSerializer for values: supports rich domain object payloads.
 * - acks=all + min.insync.replicas (set on broker) = strongest durability guarantee.
 * - Idempotent producer prevents duplicate messages on retries.
 * - Compression reduces network I/O and broker storage at the cost of CPU.
 */
@Configuration
public class KafkaProducerConfig {
 
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
 
    /**
     * Builds the raw producer configuration map.
     * Separating this into its own method makes it easy to override in tests
     * via @TestConfiguration + EmbeddedKafka.
     */
    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
 
        // ----- Connection -----
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
 
        // ----- Serialization -----
        // String key → predictable, human-readable, hashable for partition routing
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // JSON value → rich domain objects without a schema registry dependency
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        // Embed the Java type header so consumers can deserialize without guessing
        props.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, true);
 
        // ----- Durability -----
        // all = leader + every in-sync replica must acknowledge before ack is returned
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        // Idempotent producer: broker deduplicates retries using a sequence number
        // This also implicitly sets max.in.flight.requests.per.connection = 5
        // and retries = Integer.MAX_VALUE
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
 
        // ----- Throughput & Latency -----
        // Batch up to 64 KB before sending (reduces per-record overhead)
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 65536);
        // Wait up to 10 ms to fill a batch (trading latency for throughput)
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        // LZ4 compression: best balance of speed vs. compression ratio for most workloads
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
        // Buffer size before backpressure is applied to the calling thread
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); // 32 MB
 
        // ----- Retry / Timeout -----
        // How long send() will block before throwing a TimeoutException
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 5000);
        // Total time a delivery may be attempted (including retries)
        props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120000);
        // Backoff between individual retry attempts
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 500);
 
        return props;
    }
 
    /**
     * ProducerFactory is the Spring abstraction over Kafka's KafkaProducer.
     * DefaultKafkaProducerFactory creates a single shared producer by default,
     * which is thread-safe and the correct default for most applications.
     */
    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }
 
    /**
     * KafkaTemplate is the primary send API in Spring Kafka.
     * It wraps ProducerFactory and provides synchronous/asynchronous send methods,
     * transaction support, and result callback wiring.
     */
    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        KafkaTemplate<String, Object> template = new KafkaTemplate<>(producerFactory());
        // Globally observe send results — log failures at WARN level
        template.setObservationEnabled(true);
        return template;
    }
}

package com.example.kafka.config;
 
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.support.ExponentialBackOffWithMaxRetries;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.kafka.support.serializer.JsonDeserializer;
 
import java.util.HashMap;
import java.util.Map;
 
/**
 * Production-ready Kafka Consumer configuration.
 *
 * Design decisions:
 * - ErrorHandlingDeserializer wraps JsonDeserializer so a single malformed
 *   message does not poison the entire partition.
 * - Manual offset commit (MANUAL_IMMEDIATE) gives the application full control
 *   over at-least-once delivery semantics.
 * - DefaultErrorHandler with exponential backoff retries transient failures
 *   (e.g., database unavailable) before sending to a Dead Letter Topic (DLT).
 * - trusted.packages whitelist prevents Remote Code Execution during
 *   type-header-guided deserialization (see Section 5).
 */
@Configuration
public class KafkaConsumerConfig {
 
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
 
    /**
     * The consumer group ID should identify the *service*, not the instance.
     * All instances of the same microservice share this ID so Kafka distributes
     * partitions across them rather than duplicating every message to each instance.
     */
    @Value("${spring.kafka.consumer.group-id:order-service-group}")
    private String groupId;
 
    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
 
        // ----- Connection -----
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
 
        // ----- Serialization -----
        // ErrorHandlingDeserializer: on deserialization failure, the record is
        // forwarded to the error handler instead of throwing an unrecoverable exception
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                ErrorHandlingDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                ErrorHandlingDeserializer.class);
        // Delegate deserializers for the wrapped types
        props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS,
                StringDeserializer.class);
        props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS,
                JsonDeserializer.class);
 
        // ----- Security: Trusted Packages -----
        // ONLY deserialize types from these packages (see Section 5 for rationale).
        // NEVER use "*" in production.
        props.put(JsonDeserializer.TRUSTED_PACKAGES,
                "com.example.kafka.dto,com.example.shared.events");
        // Optionally map incoming type headers to local class names (useful when
        // producer and consumer live in different Maven modules / class names differ)
        props.put(JsonDeserializer.TYPE_MAPPINGS,
                "orderEvent:com.example.kafka.dto.OrderEvent," +
                "paymentEvent:com.example.kafka.dto.PaymentEvent");
 
        // ----- Offset Management -----
        // "earliest" on first join (no committed offset) — process from the beginning
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // Disable auto-commit: we commit manually after successful processing
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
 
        // ----- Fetch Tuning -----
        // Minimum data fetched per poll — reduces round-trips under low throughput
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1);
        // Maximum wait time if fetch.min.bytes is not met
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);
        // Maximum records returned per poll() call — tune based on processing time
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
        // Must process MAX_POLL_RECORDS within this window or the coordinator
        // considers the consumer dead and triggers a rebalance
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000); // 5 min
 
        return props;
    }
 
    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }
 
    /**
     * DefaultErrorHandler replaces the deprecated SeekToCurrentErrorHandler.
     *
     * Backoff strategy:
     *   Attempt 1 → immediate
     *   Attempt 2 → 1s
     *   Attempt 3 → 2s
     *   Attempt 4 → 4s
     *   Attempt 5 → 8s   ← maxRetries
     *   → publish to Dead Letter Topic (DLT)
     *
     * The DLT topic name follows the convention: <original-topic>.DLT
     * Requires a DeadLetterPublishingRecoverer bean (wired separately or inline).
     */
    @Bean
    public DefaultErrorHandler kafkaErrorHandler() {
        // Exponential backoff: initial interval 1s, multiplier 2x, max 5 retries
        ExponentialBackOffWithMaxRetries backOff =
                new ExponentialBackOffWithMaxRetries(5);
        backOff.setInitialInterval(1_000L);
        backOff.setMultiplier(2.0);
        backOff.setMaxInterval(30_000L); // cap at 30 seconds
 
        DefaultErrorHandler errorHandler = new DefaultErrorHandler(backOff);
 
        // Do NOT retry on these exceptions — they indicate a programming error,
        // not a transient infrastructure failure
        errorHandler.addNotRetryableExceptions(
                IllegalArgumentException.class,
                NullPointerException.class,
                ClassCastException.class
        );
 
        // Log each retry attempt
        errorHandler.setRetryListeners((record, ex, deliveryAttempt) ->
                System.err.printf("[KafkaRetry] topic=%s partition=%d offset=%d " +
                        "attempt=%d error=%s%n",
                        record.topic(), record.partition(), record.offset(),
                        deliveryAttempt, ex.getMessage())
        );
 
        return errorHandler;
    }
 
    /**
     * ConcurrentKafkaListenerContainerFactory creates the listener containers
     * that back @KafkaListener methods.
     *
     * concurrency = number of threads per listener, each owning one or more partitions.
     * Set this to the number of partitions (or a divisor of it) for optimal assignment.
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object>
            kafkaListenerContainerFactory() {
 
        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
 
        factory.setConsumerFactory(consumerFactory());
        factory.setCommonErrorHandler(kafkaErrorHandler());
 
        // Concurrency: 3 threads ↔ 3 partitions (adjust to match your topic)
        factory.setConcurrency(3);
 
        // MANUAL_IMMEDIATE: ack() must be called explicitly in the @KafkaListener
        // method. The offset is committed to Kafka as soon as ack() is called.
        factory.getContainerProperties()
               .setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
 
        // Enable batch listening — receive a List<ConsumerRecord> per poll
        // (comment out for record-by-record processing)
        // factory.setBatchListener(true);
 
        // Micrometer observation support for distributed tracing
        factory.getContainerProperties().setObservationEnabled(true);
 
        return factory;
    }
}

package com.example.kafka.consumer;
 
import com.example.kafka.dto.OrderEvent;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
 
@Slf4j
@Component
@RequiredArgsConstructor
public class OrderEventConsumer {
 
    private final OrderProcessingService orderProcessingService;
 
    /**
     * groupId is inherited from spring.kafka.consumer.group-id (or the bean default).
     * topics can be a SpEL expression: topics = "${app.kafka.topics.orders}"
     */
    @KafkaListener(
            topics = "${app.kafka.topics.orders:order-events}",
            containerFactory = "kafkaListenerContainerFactory"
    )
    public void handleOrderEvent(
            ConsumerRecord<String, OrderEvent> record,
            Acknowledgment ack) {
 
        log.info("Received order event: key={} partition={} offset={}",
                record.key(), record.partition(), record.offset());
 
        try {
            orderProcessingService.process(record.value());
            // Only commit the offset after successful processing
            ack.acknowledge();
        } catch (Exception e) {
            // Do NOT ack — DefaultErrorHandler will retry according to backoff policy
            log.error("Failed to process order event at offset {}: {}",
                    record.offset(), e.getMessage());
            throw e; // re-throw so the error handler can intercept
        }
    }
}

// Consumer for operational processing
@KafkaListener(
    topics = "order-events",
    groupId = "order-service-group",
    containerFactory = "kafkaListenerContainerFactory"
)
public void processOrder(ConsumerRecord<String, OrderEvent> record, Acknowledgment ack) {
    // ... business logic
}
 
// Consumer for audit trail — same topic, different group
@KafkaListener(
    topics = "order-events",
    groupId = "audit-service-group",
    containerFactory = "kafkaListenerContainerFactory"
)
public void auditOrder(ConsumerRecord<String, OrderEvent> record, Acknowledgment ack) {
    // ... write to audit log
}

# DANGEROUS — allows any class from any package
spring.kafka.consumer.properties.spring.json.trusted.packages=*
 
# CORRECT — whitelist only the packages your application owns
spring.kafka.consumer.properties.spring.json.trusted.packages=\
  com.example.kafka.dto,\
  com.example.shared.events

props.put(JsonDeserializer.TRUSTED_PACKAGES,
        "com.example.kafka.dto,com.example.shared.events");

// Producer side: publishes with header __TypeId__ = "orderEvent"
props.put(JsonSerializer.TYPE_MAPPINGS,
        "orderEvent:com.example.producer.events.OrderEvent");
 
// Consumer side: maps "orderEvent" to its own local DTO
props.put(JsonDeserializer.TYPE_MAPPINGS,
        "orderEvent:com.example.consumer.dto.OrderEvent");