@RequiredArgsConstructor
public class KafkaService {

    private final KafkaTemplate<String, String> kafkaTemplate;

    @Async
    public void sendMessage(String topicName, String message) {

        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topicName, message);

        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() { //  비동기 형태로 브로커로 전송된 메시지의 정상적인 적재 여부를 파악한다.

            @Override
            public void onSuccess(SendResult<String, String> result) {}

            public void onFailure(Throwable ex) {
                log.error("Unable to topic : {}, send message : {} exception : {}", topicName, message, ex.getMessage());
            }
        });
    }
}

@Configuration
@RequiredArgsConstructor
public class KafkaProducerConfig {
    
    private final KafkaProperties kafkaProperties;

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.ACKS_CONFIG, "all");
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

}

@KafkaListener(topics = "test", groupId = "test-group-00")
public void recordListener(ConsumerRecord<String, String> record) {
    log.info(record.toString());
    // 기본적인 리스너선언 방식으로, poll()이 호출되어 가져온 레코드들을 차례대로 개별 레코드의 메시지 값을 파라미터로 받게 된다.
    // 파라미터로 컨슈머 레코드를 받기 때문에 메시지 키, 메시지 값에 대한 처리를 이 메서드 안에서 수행하면 된다.
}

@KafkaListener(topics = "test", groupId = "test-group-01")
public void singleTopicListener(String messageValue) {
    log.info(messageValue);
    // 메시지 값을 파라미터로 받는 리스너
}

@KafkaListener(topics = "test", groupId = "test-group-02", properties = {"max.poll.interval.ms:60000", "auto.offset.reset:earliest"})
public void singleTopicWithPropertiesListener(String messageValue) {
    log.info(messageValue);
    // 별도의 프로퍼티 옵션값을 선언해주고 싶을 때 사용한다.
}

@KafkaListener(topics = "test", groupId = "test-group-03", concurrency = "3")
public void concurrentTopicListener(String messageValue) {
    log.info(messageValue);
    // 2개 이상의 카프카 컨슈머 스레드를 실행하고 싶을 때 concurrency 옵션을 활용할 수 있다.
    // concurrency값 만큼 컨슈머 스레드를 생성하여 병렬처리 한다.
}

@KafkaListener(topicPartitions = {
    @TopicPartition(topic = "test01", partitions = {"0", "1"}),
    @TopicPartition(topic = "test02", partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "3")),
})
public void listenSpecificPartition(ConsumerRecord<String, String> record) {
    log.info(record.toString());
    // 특정 토픽의 특정 파티션만 구독하고 때 `topicPartitions` 파라미터를 사용한다.
    // `PartitionOffset` 어노테치션을 활용하면 특정 파티션의 특정 오프셋까지 지정할 수 있다.
    // 이 경우에는 그룹 아이디에 관계없이 항상 설정한 오프셋의 데이터부터 가져온다.
}

@KafkaListener(topics = "test", groupId = "test-group-00")
public void batchListener(ConsumerRecords<String, String> records) {
    records.forEach(record -> log.info(record.toString()));
    // 컨슈머 레코드의 묶음(ConsumerRecords)을 파라미터로 받는다.
    // 카프카 클라이언트 라이브러리에서 poll() 메서드로 리턴받은 ConsumerRecords를 리턴받아 사용하는 방식과 같다.
}

@KafkaListener(topics = "test", groupId = "test-group-01")
public void singleTopicListener(List<String> list) {
    list.forEach(recordValue -> log.info(recordValue));
    // 메시지 값을 List형태로 받는다.
}

@KafkaListener(topics = "test", groupId = "test-group-02", concurrency = "3")
public void concurrentTopicListener(ConsumerRecords<String, String> records) {
    records.forEach(record -> log.info(record.toString()));
    // 2개 이상의 컨슈머 스레드로 배치 리스너를 운영할 경우에 concurrency 옵션을 함께 선언하여 사용하면 된다.
}

@KafkaListener(topics = "test", groupId = "test-group-00")
public void commitListener(ConsumerRecords<String, String> records, Acknowledgment ack) {
    records.forEach(record -> log.info(record.toString()));
    ack.acknowledge();
    // AckMode를 MANUAL또는 MANUAL_IMMEDIATE로 사용할 경우 수동 커밋을 하기 위해 파라미터로 Acknowledgment 인스턴스를 받아야 한다.
    // acknowledge() 메서드를 호출함으로써 커밋을 수행할 수 있다.
}

@KafkaListener(topics = "test", groupId = "test-group-01")
public void consumerCommitTopicListener(ConsumerRecords<String, String> records, Consumer<String, String> consumer) {
    records.forEach(record -> log.info(record.toString()));
    consumer.commitAsync();
    // 동기/비동기 커밋을 사용하고 싶다면 컨슈머 인스턴스를 파라미터로 받아서 사용하면 된다.
    // consumer 인스턴스의 commitSync(), commitAsync() 메서드를 호출하면 사용자가 원하는 타이밍에 커밋할 수 있도록 로직을 추가할 수 있다.
    // 다만 리스너가 커밋을 하지 않도록 AckMode는 MANUAL, MANUAL_IMMEDIATE로 설정해야 한다.
}

@Configuration
public class ListenerContainerConfiguration {
    
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> customContainerFactory() {
        
        Map<String, Object> props = new HashMap<>();
        props.put(Consumerconfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:8080");
        props.put(Consumerconfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(Consumerconfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        
        DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props);
        
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();  //  리스너 컨테이너를 만들기 위해 사용
        factory.getContainerProperties().setConsumerRebalanceListener(new ConsumerAwareRebalanceListener() {
            // 리밸런스 리스너를 선언하기 위해 setConsumerRebalanceListener 메서드를 호출한다.
            
            @Override
            public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
                // 커밋이 되기 전 리밸런스가 발생했을 때 호출되는 메서드
            }
            
            @Override
            public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
                // 커밋이 일어난 이후 리밸런스가 발생했을 때 호출되는 메서드
            }
            
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                // 리밸런싱이 끝나서 파티션 소유권이 할당되고 나면 호출되는 메서드
            }
            
            @Override
            public void onPartitionsLost(Collection<TopicPartition> partitions) {
                
            }
            
        });
        
        factory.setBatchListener(false);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.RECORD);
        factory.setConsumerFactory(cf);
        
        return factory;
    }
    
}

@KafkaListener(topics = "test", groupId = "test-group", containerFactory = "customContainerFactory")
public void customListener(String data) {
    log.info(data);
    // customContainerFactory 옵션을 커스텀 컨테이너 팩토리로 설정하여 사용한다.
}