5. 스프링 카프카 동작 원리

public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageListenerContainer<K, V> {
    private final List<KafkaMessageListenerContainer<K, V>> containers = new ArrayList<>();
    // 생략

@FunctionalInterface
public interface BatchMessageListener<K, V> extends GenericMessageListener<List<ConsumerRecord<K, V>>> {
    // 생략
}

@FunctionalInterface
public interface MessageListener<K, V> extends GenericMessageListener<ConsumerRecord<K, V>> {

}

// 어노테이션 생략
public class KafkaConsumerConfig {
  // 생략
  @Bean
  public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, Chat>> kafkaListenerContainerFactory() {
    final ConcurrentKafkaListenerContainerFactory<String, Chat> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setBatchListener(true); // 여기
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);

    return factory;
  }
  // 생략
}

public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMessageListenerContainer<K, V>, K, V>
		implements KafkaListenerContainerFactory<C>, ApplicationEventPublisherAware, InitializingBean,
{
  @Override
  public C createListenerContainer(KafkaListenerEndpoint endpoint) {
    C instance = createContainerInstance(endpoint);
    JavaUtils.INSTANCE
            .acceptIfNotNull(endpoint.getId(), instance::setBeanName)
            .acceptIfNotNull(endpoint.getMainListenerId(), instance::setMainListenerId);
    if (endpoint instanceof AbstractKafkaListenerEndpoint) {
      configureEndpoint((AbstractKafkaListenerEndpoint<K, V>) endpoint);
    }

    if (Boolean.TRUE.equals(endpoint.getBatchListener())) {
      endpoint.setupListenerContainer(instance, this.batchMessageConverter);
    }
    else {
      endpoint.setupListenerContainer(instance, this.recordMessageConverter);
    }
    initializeContainer(instance, endpoint);
    customizeContainer(instance);
    return instance;
  }
}

public abstract class AbstractKafkaListenerEndpoint<K, V>
		implements KafkaListenerEndpoint, BeanFactoryAware, InitializingBean
{
    // 생략
    private void setupMessageListener(MessageListenerContainer container,
                                      @Nullable MessageConverter messageConverter) {

      MessagingMessageListenerAdapter<K, V> adapter = createMessageListener(container, messageConverter);
      JavaUtils.INSTANCE
              .acceptIfNotNull(this.replyHeadersConfigurer, adapter::setReplyHeadersConfigurer)
              .acceptIfNotNull(this.correlationHeaderName, adapter::setCorrelationHeaderName);
      adapter.setSplitIterables(this.splitIterables);
      Object messageListener = adapter;
      boolean isBatchListener = isBatchListener();
      Assert.state(messageListener != null,
              () -> "Endpoint [" + this + "] must provide a non null message listener");
      if (this.recordFilterStrategy != null) {
        if (isBatchListener) {
          if (((MessagingMessageListenerAdapter<K, V>) messageListener).isConsumerRecords()) {
            this.logger.warn(() -> "Filter strategy ignored when consuming 'ConsumerRecords' instead of a List"
                    + (this.id != null ? " id: " + this.id : ""));
          }
          else {
            messageListener = new FilteringBatchMessageListenerAdapter<>(
                    (BatchMessageListener<K, V>) messageListener, this.recordFilterStrategy, this.ackDiscarded);
          }
        }
        else {
          messageListener = new FilteringMessageListenerAdapter<>((MessageListener<K, V>) messageListener,
                  this.recordFilterStrategy, this.ackDiscarded);
        }
      }
      container.setupMessageListener(messageListener);
    }
}

public class BatchMessagingMessageConverter implements BatchMessageConverter {
  @Override // NOSONAR
  public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, @Nullable Acknowledgment acknowledgment,
                              Consumer<?, ?> consumer, Type type) {

    KafkaMessageHeaders kafkaMessageHeaders = new KafkaMessageHeaders(this.generateMessageId,
            this.generateTimestamp);

    // 생략

    for (ConsumerRecord<?, ?> record : records) {
      payloads.add(obtainPayload(type, record, conversionFailures));
      keys.add(record.key());
      topics.add(record.topic());
      partitions.add(record.partition());
      offsets.add(record.offset());
      if (record.timestampType() != null) {
        timestampTypes.add(record.timestampType().name());
      }
      timestamps.add(record.timestamp());
      if (this.headerMapper != null && record.headers() != null) {
        Map<String, Object> converted = new HashMap<>();
        this.headerMapper.toHeaders(record.headers(), converted);
        convertedHeaders.add(converted);
        Object object = converted.get(KafkaHeaders.LISTENER_INFO);
        if (object instanceof String) {
          info = (String) object;
        }
      }
      // 생략
    return MessageBuilder.createMessage(payloads, kafkaMessageHeaders);
  }
}