@Slf4j
@RestController
@RequestMapping("/api")
public class KafkaProducerController {
    // kafka producer를 위한 KafkaTemplate를 지정한다.
    private final KafkaTemplate<String, Object> kafkaProducerTemplate;

    @Value("${kafka.topic-login-request}")
    public String TOPIC_LOGIN_REQUEST;

    @Value("${kafka.topic-login-response}")
    public String TOPIC_LOGIN_RESPONSE;

    ...

    // 임시로 PostMapping하였다.
    // 역할1. userId와 userPw를 담은 RequestLoginDTO가 도착하면 이를 kafka의 특정 파티션으로 전송
    @PostMapping("login")
    public ResponseEntity<?> produceMessageWithRequestLoginDTO(@RequestBody RequestLoginDTO requestLoginDTO) {

        // 이부분이 중요하다!!!!
        // userId를 Kafka의 토픽에 key로 던져주면서 특정 파티션에 들어가도록 설정한다
        // (던져주면 partition = HASH(useriD) mod (파티션개수) 로 수행할듯)
        ListenableFuture<SendResult<String, Object>> future = kafkaProducerTemplate.send(TOPIC_LOGIN, requestLoginDTO.getUserId(), requestLoginDTO);

        // callback
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                log.error("Unable to send message: {}", ex.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                log.info("Sent message with key: {}, offset: {}, partition: {}", requestLoginDTO.getUserId(), result.getRecordMetadata().offset(), result.getRecordMetadata().partition());
            }
        });
        return ResponseEntity.ok(requestLoginDTO);
    }
}

@Slf4j
@Component
public class MessageListener {

    ...

    // 로그인
    @KafkaListener(topics = "${kafka.topic-login-request}", containerFactory = "loginKafkaListenerContainerFactory")
    public void listenLogin(RequestLoginDTO loginDTO) {
        log.info("Receive [RequestLoginDTO] Message with userID={},userPw={}", loginDTO.getUserId(), loginDTO.getUserPw());

        Optional<User> user = userService.matchUserIdAndUserPw(loginDTO.getUserId(), loginDTO.getUserPw());

        ResponseLoginDTO responseLoginDTO = new ResponseLoginDTO();
        responseLoginDTO.setRequestUserID(loginDTO.getUserId());

        if (user.isPresent()) {
            responseLoginDTO.setIsAccept(true);
            responseLoginDTO.setUser(user.get());
        } else {
            responseLoginDTO.setIsAccept(false);
            responseLoginDTO.setUser(new User());
        }

        // Kafka 메세지 전송/비동기 처리를 위한 ListenableFuture 사용
        ListenableFuture<SendResult<String, Object>> future = kafkaProducerTemplate.send(TOPIC_LOGIN_RESPONSE, loginDTO.getUserId(), responseLoginDTO);

        // 메시지 비동기 callback 처리
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                log.error("Fail to send message to broker: {}", ex.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                log.info("Send message with offset: {}, partition: {}", result.getRecordMetadata().offset(), result.getRecordMetadata().partition());
            }
        });
    }
}

// 역할1. Kafka 토픽 할당
@Configuration
public class KafkaTopicConfig {
    @Autowired
    private KafkaAdmin kafkaAdmin;

    @Value("${kafka.topic-login-request}")
    public String TOPIC_LOGIN_REQUEST;

    @Value("${kafka.topic-login-response}")
    public String TOPIC_LOGIN_RESPONSE;

    // 계산 잘 해야한다. Partition 개수 >= Group내 Conusmer 개수.
    // 생성하고자 하는 Conumser=2 Partition은 2이기에, 각각 conusmer에게 leader-partition 매칭가능
    private NewTopic generateTopic(String topicName) {
      return TopicBuilder.name(topicName)
        .partitions(2) // 파티션 할당 개수
        .replicas(3) // broker 3대에 할당
        .build(); // 즉, 토픽은 총 2개의 leader-partition, 2개의 follow-partition 보유
    }

    @PostConstruct
    public void init() {
      kafkaAdmin.createOrModifyTopics(generateTopic(TOPIC_LOGIN_REQUEST));
      kafkaAdmin.createOrModifyTopics(generateTopic(TOPIC_LOGIN_RESPONSE));
    }
}

@Configuration
@EnableKafka
public class KafkaProducerConfig {

    @Value("${kafka.bootstrap-servers}")
    private String bootstrapServer;

    private ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        // Kafka Broker 엔드 포인트 할당(8097, 8098, 8099)
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
        // Kafka에 전송될 키는 스트링으로 설정
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // Kafka에 전송될 값을 직렬화 하는 방법을 지정
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

        return new DefaultKafkaProducerFactory<>(configProps);
    }

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

}

// 역할1. Kafka연결
// 역할2. 메세지를 받아서 우리가 원하는 타입인 RequestLoginDTO로 변환
@Slf4j
@EnableKafka
@Configuration
public class KafkaConsumerConfig {

    @Value("${kafka.bootstrap-servers}")
    private String bootstrapServer;

    ...

    // login consumer 객체 변환
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, RequestLoginDTO> loginKafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, RequestLoginDTO> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(loginConsumerFactory("chatServerGroup"));
        return factory;
    }

    // login consumer 객체 변환
    public ConsumerFactory<String, RequestLoginDTO> loginConsumerFactory(String groupId) {

        // 앞서 producer이 json직렬화로 보냈으니, 마찬가지로 jsonDeserialize
        JsonDeserializer<RequestLoginDTO> deserializer = new JsonDeserializer<>(RequestLoginDTO.class);
        deserializer.setRemoveTypeHeaders(false);
        // 모든 패키지 신뢰
        deserializer.addTrustedPackages("*");
        deserializer.setUseTypeMapperForKey(true);

        // 객체변환 설정
        ImmutableMap<String, Object> config = ImmutableMap.<String, Object>builder()
                .put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer)
                .put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class)
                .put(ConsumerConfig.GROUP_ID_CONFIG, groupId)
                // 역직렬화 로직
                .put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, deserializer)
                .build();

        return new DefaultKafkaConsumerFactory<>(config, new StringDeserializer(), deserializer);
    }
}