Consuming Events with KafkaListener

@Component
@RequiredArgsConstructor
@Slf4j
class OrderPlacedListener {

    private final PaymentService paymentService;

    @KafkaListener(topics = "order-placed", groupId = "payment-service", concurrency = "3")
    void onOrderPlaced(OrderPlacedEvent event,
                       @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                       @Header(KafkaHeaders.OFFSET) long offset) {
        log.info("OrderPlaced {} (p{}@{}) amount {}", event.orderId(), partition, offset, event.totalAmount());
        paymentService.startPayment(event);
    }
}

spring:
  kafka:
    bootstrap-servers: ${KAFKA_SERVERS:localhost:9092}
    consumer:
      group-id: payment-service
      auto-offset-reset: earliest     # new group: start from the oldest record
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
        spring.json.value.default.type: com.shopease.payment.events.OrderPlacedEvent
        spring.json.trusted.packages: "com.shopease.*"

• concurrency = "3" starts 3 consumer threads, one per partition. More threads than partitions would sit idle.
• ErrorHandlingDeserializer wraps the JSON deserializer so one broken message ("poison pill") does not crash the listener in a loop.
• Spring commits the offset after the method returns normally (ack mode BATCH by default).
• Each service has its own copy of the event class in its own package; they share the JSON contract, not a jar.