Implementing Messaging with Kafka

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
</dependency>

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
      retries: 5
      properties:
        enable.idempotence: true
        compression.type: lz4

spring:
  kafka:
    consumer:
      group-id: orders-consumer
      auto-offset-reset: earliest
      enable-auto-commit: false
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "*"
    listener:
      ack-mode: MANUAL_IMMEDIATE

@Service
public class OrderPublisher {
  private final KafkaTemplate<String, OrderPlaced> kafka;
  public OrderPublisher(KafkaTemplate<String, OrderPlaced> kafka) { this.kafka = kafka; }
  public void publish(OrderPlaced evt) {
    kafka.send("orders.placed", String.valueOf(evt.orderId()), evt)
         .whenComplete((res, ex) -> { if (ex != null) log.error("send failed", ex); });
  }
}

@Component
public class OrderConsumer {
  @KafkaListener(topics = "orders.placed", groupId = "orders-consumer", concurrency = "3")
  public void handle(ConsumerRecord<String, OrderPlaced> rec, Acknowledgment ack) {
    process(rec.value());
    ack.acknowledge();
  }
}

@Bean
DefaultErrorHandler errorHandler(KafkaTemplate<String,Object> template) {
  DeadLetterPublishingRecoverer dlr = new DeadLetterPublishingRecoverer(template);
  return new DefaultErrorHandler(dlr, new FixedBackOff(1000L, 3));
}

@KafkaListener(topics="orders.placed")
public void handle(@Payload OrderPlaced evt,
                   @Header(KafkaHeaders.RECEIVED_KEY) String key,
                   @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts,
                   @Headers Map<String,Object> headers) { /* … */ }

@Configuration
@EnableKafkaStreams
public class StreamsConfig {
  @Bean
  KStream<String, OrderPlaced> pipeline(StreamsBuilder b) {
    KStream<String,OrderPlaced> s = b.stream("orders.placed");
    s.filter((k,v) -> v.total().compareTo(BigDecimal.valueOf(1000)) > 0)
     .to("orders.high-value");
    return s;
  }
}