content: package com.food.order.system.restaurant.messaging.listener.kafka;
  
  
  import com.food.order.system.kafka.consumer.KafkaConsumer;
  import com.food.order.system.kafka.order.avro.model.RestaurantApprovalRequestAvroModel;
  import com.food.order.system.restaurant.messaging.mapper.RestaurantMessagingDataMapper;
  import com.food.order.system.restaurant.domain.service.ports.input.message.listener.RestaurantApprovalRequestMessageListener;
  import lombok.extern.slf4j.Slf4j;
  import org.springframework.kafka.annotation.KafkaListener;
  import org.springframework.kafka.support.KafkaHeaders;
  import org.springframework.messaging.handler.annotation.Header;
  import org.springframework.messaging.handler.annotation.Payload;
  import org.springframework.stereotype.Service;
  
  import java.util.List;
  
  @Slf4j
  @Service
  public class RestaurantApprovalRequestKafkaListener implements KafkaConsumer<RestaurantApprovalRequestAvroModel> {
  
      private final RestaurantApprovalRequestMessageListener restaurantApprovalRequestMessageListener;
      private final RestaurantMessagingDataMapper restaurantMessagingDataMapper;
  
      public RestaurantApprovalRequestKafkaListener(RestaurantApprovalRequestMessageListener restaurantApprovalRequestMessageListener,
                                                    RestaurantMessagingDataMapper restaurantMessagingDataMapper) {
          this.restaurantApprovalRequestMessageListener = restaurantApprovalRequestMessageListener;
          this.restaurantMessagingDataMapper = restaurantMessagingDataMapper;
      }
  
      @Override
      @KafkaListener(id = "${kafka-consumer-config.restaurant-approval-consumer-group-id}",
              topics = "${restaurant-service.restaurant-approval-request-topic-name}")
      public void receive(@Payload List<RestaurantApprovalRequestAvroModel> messages,
                          @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) List<String> keys,
                          @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                          @Header(KafkaHeaders.OFFSET) List<Long> offsets) {
          log.info("{} number of orders approval requests received with keys {}, partitions {} and offsets {}" +
                          ", sending for restaurant approval",
                  messages.size(),
                  keys.toString(),
                  partitions.toString(),
                  offsets.toString());
  
          messages.forEach(restaurantApprovalRequestAvroModel -> {
              log.info("Processing order approval for order id: {}", restaurantApprovalRequestAvroModel.getOrderId());
              restaurantApprovalRequestMessageListener.approveOrder(restaurantMessagingDataMapper.
                      restaurantApprovalRequestAvroModelToRestaurantApproval(restaurantApprovalRequestAvroModel));
          });
      }
  
  }
  