Some Random Notes

Gem 2

2 min read

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
 
@SpringBootApplication
public class KafkaDemoApplication {
    public static void main(String[] args) {
        SpringApplication.run(KafkaDemoApplication.class, args);
    }
}
 
// ==========================================
// 1. CODE EXAMPLES (Template & Listener)
// ==========================================
@RestController
@RequestMapping("/api/messages")
class MessageController {
 
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
 
    @PostMapping("/send")
    public String sendMessage(@RequestParam String message) {
        // Using KafkaTemplate to send a message to the "orders-topic"
        kafkaTemplate.send("orders-topic", message);
        return "Message sent to Kafka successfully!";
    }
}
 
class MessageConsumer {
 
    // Using @KafkaListener to automatically consume messages from "orders-topic"
    @KafkaListener(topics = "orders-topic", groupId = "order-processing-group")
    public void listen(String message) {
        System.out.println("Received message: " + message);
    }
}
# ==========================================
# 2. CONFIGURATION PROPERTIES (application.yml)
# ==========================================
spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      # Serializes keys and values into bytes before sending to the broker
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      acks: all # Ensures maximum data durability
    consumer:
      # Deserializes bytes back into Java objects/strings upon receipt
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      group-id: order-processing-group
      auto-offset-reset: earliest # Reads from the beginning if no previous offset exists
    listener:
      # Concurrency spins up multiple Kafka message listener threads to process partitions in parallel
      concurrency: 3 
// ==========================================
// 3. ERROR HANDLING & RETRY MECHANISMS
// ==========================================
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.retrytopic.RetryableTopic;
import org.springframework.kafka.annotation.DltHandler;
import org.springframework.util.backoff.FixedBackOff;
 
@Configuration
class KafkaErrorHandlerConfig {
 
    // Automatically sets up Non-Blocking Retries and a Dead Letter Topic (DLT)
    @RetryableTopic(
        attempts = "3", 
        backoff = @org.springframework.kafka.retrytopic.BackOff(delay = 2000), // 2 seconds between retries
        dltTopicSuffix = "-dlt"
    )
    @KafkaListener(topics = "payment-topic", groupId = "payment-group")
    public void processPayment(String paymentMessage) {
        if (paymentMessage.contains("fail")) {
            throw new RuntimeException("Simulated payment processing failure");
        }
        System.out.println("Payment processed: " + paymentMessage);
    }
 
    // This method handles messages that failed all 3 attempts
    @DltHandler
    public void handleDeadLetterMessage(String failedMessage) {
        System.err.println("Message moved to DLT after maximum retries: " + failedMessage);
    }
}

If you want to tailor this implementation, let me know:

  • What data format are you passing? (JSON, String, Avro?)
  • Should we configure blocking retries (pausing the consumer partition) or non-blocking retries (forwarding to retry topics)?
  • Do you need help setting up integration tests using EmbeddedKafka?

Let me know what you’d like to explore next!

Backlinks