What is Spring Boot's messaging support with Kafka in Boot 4?

// Producer:
@Service
public class OrderEventProducer {

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

    public void publishOrderCreated(Order order) {
        OrderEvent event = new OrderEvent(order.id(), "ORDER_CREATED", Instant.now());
        kafkaTemplate.send("orders-topic", order.id(), event);
    }
}

// Consumer:
@Service
public class OrderEventConsumer {

    @KafkaListener(
        topics       = "orders-topic",
        groupId      = "order-processing-group",
        concurrency  = "3"  // 3 consumer threads
    )
    public void handleOrderEvent(
            @Payload OrderEvent event,
            @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
            Acknowledgment ack) {
        try {
            processEvent(event);
            ack.acknowledge();  // manual acknowledgement
        } catch (Exception e) {
            log.error("Failed to process event: {}", event, e);
            // send to DLT (dead letter topic)
        }
    }
}

# application.yml: Kafka configuration
spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: order-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "com.example.events"
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer