kafka-patterns

Apache Kafka message queue

Source:https://kafka.apache.org/documentation/
Spring Kafka：https://docs.spring.io/spring-kafka/reference/

Capability Boundaries

✅ Strong Suits

1. Producer configuration — acks (retry/partition)/retries (batchSize/linger)/idempotence
2. Consumer configuration — auto-offset-reset/enable-auto-commit/manual commit
3. Idempotence guarantee — producer enable.idempotence + acks=all + consumer business deduplication
4. Spring Kafka — @KafkaListener / @RetryableTopic / @DltHandler
5. DLT dead letter — DefaultErrorHandler + DeadLetterPublishingRecoverer
6. Ordered messages — Same Key sent to the same partition
7. Transactional messages — @Transactional + KafkaTransactionManager

❌ Out of Scope

1. Kafka Streams → stream processing, not in the message queue scope
2. Kafka Connect → data pipeline, not an LLM code issue
3. Schema Registry → serialization scheme selection

Producer acks selection decision

acks value	Reliability	Latency	Use case
acks=all(-1)	⭐⭐⭐ Highest	High	Core business (orders/payments/transactions) ✅
acks=1	⭐⭐ Medium	Medium	Default value, general business
acks=0	⭐ Lowest	Low	Logs/monitoring/non-critical metrics

Core business must: acks=all + min.insync.replicas=2(Broker configuration) + replication.factor=3

LLM's most common mistakes

#	Mistake	Correct approach
1	acks=0 or acks=1 loses messages	Core business acks=all + min.insync.replicas=2
2	enable.auto.commit=true (auto-commit offset)	Manual commit, acknowledge after processing
3	Consumer does not handle duplicate consumption	Business-layer idempotence (unique key + database deduplication)
4	No DLT dead letter configured, exception messages retried indefinitely	DefaultErrorHandler + DeadLetterPublishingRecoverer
5	All messages sent to the same partition	Route to partitions using Key, same Key guarantees order
6	Synchronous send producer.send().get()	Asynchronous send + callback to handle exceptions
7	enable.idempotence=true not configured	Idempotent production prevents duplicate messages caused by retries
8	Exceptions in the listener are not handled	Configure ErrorHandler, exception→DLT

Core patterns

Pattern 1: Producer configuration (core business)

@Bean
public ProducerFactory<String, Object> producerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    config.put(ProducerConfig.ACKS_CONFIG, "all");                    // ✅ 所有副本确认
    config.put(ProducerConfig.RETRIES_CONFIG, 3);                      // 重试3次
    config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);        // ✅ 幂等
    config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
    config.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);    // 投递超时
    config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "zstd");        // 压缩
    config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
    return new DefaultKafkaProducerFactory<>(config);
}

Pattern 2: Consumer manual commit + DLT

@Bean
public ConsumerFactory<String, Object> consumerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ConsumerConfig.GROUP_ID_CONFIG, "order-group");
    config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");   // 从最早未消费开始
    config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);       // ✅ 手动提交
    config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
    config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    return new DefaultKafkaConsumerFactory<>(config);
}

// ✅ DLT 配置
@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory(
        ConsumerFactory<String, Object> cf, KafkaTemplate<String, Object> kt) {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(cf);
    // 重试3次，每次间隔1秒，之后进DLT
    factory.setCommonErrorHandler(new DefaultErrorHandler(
        new DeadLetterPublishingRecoverer(kt),
        new FixedBackOff(1000L, 3L)));
    return factory;
}

Pattern 3: Spring Kafka listener + DLT

// 方式一：直接 @KafkaListener + DLT ErrorHandler
@KafkaListener(topics = "order-topic", groupId = "order-group")
public void listen(ConsumerRecord<String, OrderEvent> record, Acknowledgment ack) {
    try {
        processOrder(record.value());
        ack.acknowledge();          // ✅ 手动提交 offset
    } catch (Exception e) {
        log.error("处理订单失败: key={}", record.key(), e);
        throw e;                    // 抛出 → ErrorHandler → 重试 → DLT
    }
}

// 方式二：@RetryableTopic 自动重试 + DLT
@RetryableTopic(
    attempts = "4",
    backoff = @Backoff(delay = 1000, multiplier = 2.0, maxDelay = 10000),
    kafkaTemplate = "kafkaTemplate")
@KafkaListener(topics = "order-topic", groupId = "order-group")
public void processOrder(OrderEvent event) {
    // 业务逻辑，失败自动重试(1s→2s→4s→...)
}

@DltHandler
public void handleDlt(OrderEvent event) {
    log.error("消息进入死信队列: {}", event);
    // 人工处理或告警
}

Pattern 4: Message sending (async + callback)

@Service
public class OrderEventPublisher {
    private final KafkaTemplate<String, Object> kafkaTemplate;

    public void publishOrderCreated(OrderEvent event) {
        String key = String.valueOf(event.getOrderId());  // ✅ 同一订单→同一分区
        ListenableFuture<SendResult<String, Object>> future =
            kafkaTemplate.send("order-topic", key, event);

        // ✅ 异步处理发送结果
        future.addCallback(result -> log.debug("发送成功: {}", result.getRecordMetadata().offset()),
            ex -> log.error("发送失败: key={}", key, ex));
    }
}

Pattern 5: Business idempotence (consumer-side deduplication)

@KafkaListener(topics = "order-topic")
public void listen(OrderEvent event) {
    // ✅ 业务幂等：用事件ID去重
    String eventId = event.getEventId();
    Boolean existed = redisTemplate.opsForValue().setIfAbsent(
        "kafka:dedup:" + eventId, "1", Duration.ofHours(24));
    if (Boolean.FALSE.equals(existed)) {
        log.debug("重复事件跳过: {}", eventId);
        return;
    }
    // 处理业务
    processOrder(event);
}

Gotchas

1. acks=all is safest but slowest — log/monitoring messages can use acks=1
2. Idempotence enable.idempotence=true requires acks=all + max.in.flight.requests.per.connection<=5
3. Manual offset commit must be in finally or Spring Acknowledgment — missed commit = duplicate consumption
4. DLT retry count is limited — messages exceeding the retry count enter the DLT Topic and require manual handling
5. Within the same group, each partition is consumed by only one consumer — Consumers > partitions → extra consumers sit idle
6. When offset starts from earliest, all historical messages are replayed — New consumer groups need to be aware of this
7. Producer and consumer serializers must match — Use JsonSerializer/JsonDeserializer instead of String
8. Retries on exceptions may cause out-of-order processing — For strict ordering scenarios, use a single partition or RabbitMQ
9. @RetryableTopic creates additional retry topics — Beware of topic count explosion
10. Compression (zstd/snappy) reduces network bandwidth — zstd is recommended (high compression ratio)
11. max.poll.records controls the number of records fetched per batch — Match processing speed to avoid rebalance
12. Large messages (>1MB) require configuring max.message.bytes — Recommended < 10MB

Data Privacy

This skill does not collect, store, or transmit any user data.