content: package com.food.order.system.messaging.publisher.kafka;
  
  import com.food.order.system.kafka.order.avro.model.PaymentRequestAvroModel;
  import com.food.order.system.kafka.producer.KafkaMessageHelper;
  import com.food.order.system.kafka.producer.service.KafkaProducer;
  import com.food.order.system.messaging.mapper.OrderMessagingDataMapper;
  import com.food.order.system.outbox.OutboxStatus;
  import com.food.order.system.config.OrderServiceConfigData;
  import com.food.order.system.outbox.model.payment.OrderPaymentEventPayload;
  import com.food.order.system.outbox.model.payment.OrderPaymentOutboxMessage;
  import com.food.order.system.ports.output.message.publisher.payment.PaymentRequestMessagePublisher;
  import lombok.RequiredArgsConstructor;
  import lombok.extern.slf4j.Slf4j;
  import org.springframework.stereotype.Component;
  
  import java.util.function.BiConsumer;
  
  
  @Slf4j
  @Component
  @RequiredArgsConstructor
  public class OrderPaymentEventKafkaPublisher implements PaymentRequestMessagePublisher {
  
      private final OrderMessagingDataMapper orderMessagingDataMapper;
      private final KafkaProducer<String, PaymentRequestAvroModel> kafkaProducer;
      private final OrderServiceConfigData orderServiceConfigData;
      private final KafkaMessageHelper kafkaMessageHelper;
  
      @Override
      public void publish(OrderPaymentOutboxMessage orderPaymentOutboxMessage,
                          BiConsumer<OrderPaymentOutboxMessage, OutboxStatus> outboxCallback) {
          OrderPaymentEventPayload orderPaymentEventPayload =
                  kafkaMessageHelper.getOrderEventPayload(orderPaymentOutboxMessage.getPayload(),
                          OrderPaymentEventPayload.class);
  
          String sagaId = orderPaymentOutboxMessage.getSagaId().toString();
  
          log.info("Received OrderPaymentOutboxMessage for order id: {} and saga id: {}",
                  orderPaymentEventPayload.getOrderId(),
                  sagaId);
  
          try {
              PaymentRequestAvroModel paymentRequestAvroModel = orderMessagingDataMapper
                      .orderPaymentEventToPaymentRequestAvroModel(sagaId, orderPaymentEventPayload);
  
              kafkaProducer.send(orderServiceConfigData.getPaymentRequestTopicName(),
                      sagaId,
                      paymentRequestAvroModel,
                      kafkaMessageHelper.getKafkaCallback(orderServiceConfigData.getPaymentRequestTopicName(),
                              paymentRequestAvroModel,
                              orderPaymentOutboxMessage,
                              outboxCallback,
                              orderPaymentEventPayload.getOrderId(),
                              "PaymentRequestAvroModel"));
  
              log.info("OrderPaymentEventPayload sent to Kafka for order id: {} and saga id: {}",
                      orderPaymentEventPayload.getOrderId(), sagaId);
          } catch (Exception e) {
             log.error("Error while sending OrderPaymentEventPayload" +
                             " to kafka with order id: {} and saga id: {}, error: {}",
                     orderPaymentEventPayload.getOrderId(), sagaId, e.getMessage());
          }
  
  
      }
  }
  