kafka Producer & Consumer 비즈니스 로직 외 설정 부분

[cotchan]

작업

• Kafka Producer, Consumer의 비즈니스 로직 처리외에, 적용해야한다고 생각한 부분들을 공부, 리서치하며 설정한 내용 정리

• 100

• 101

목차

1. Producer 추가 설정
2. Producer 예외 핸들링 적용
3. Consumer 예외 핸들링 적용

---

Producer 추가 설정

• 프로듀서 필수 옵션인 bootstrap.servers, key.serializer, value.serializer 제외하고 필요하다고 생각한 옵션인 retries 10으로 설정

retries

• 브로커로부터 에러를 받고 난 뒤 재전송을 시도하는 횟수를 지정
• 기본값은 INTEGER.MAX_VALUE
• ProducerConfig.RETRIES_CONFIG

@Configuration
public class KafkaProducerConfig {
    //...
    private Map<String, Object> producerConfigs() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        configs.put(ProducerConfig.RETRIES_CONFIG, retries);
        return configs;
    }
}

spring:
  kafka:
    producer:
      retries: 10

---

Producer 예외 핸들링 적용

• 레퍼런스 참고하여 KafkaTemplate.send() 호출 시 반환하는 ListenableFuture에 콜백함수 적용함으로써 예외 로깅 처리
  • Intro to Apache Kafka with Spring

@Slf4j
public class BuyerReviewCreatedEventListener implements AutoCloseable {

    @Value("${spring.kafka.topic.buyer-review-created}")
    private String buyerReviewCreatedTopic;

    private final EventBus eventBus;

    private final KafkaTemplate<String, BuyerReviewCreatedMessage> kafkaTemplate;

    //...

    @Subscribe
    public void handleBuyerReviewCreatedEvent(BuyerReviewCreatedEvent event) {
        Long sellerId = event.getSellerId();
        Long reviewerId = event.getReviewerId();

        ListenableFuture<SendResult<String, BuyerReviewCreatedMessage>> future =
            kafkaTemplate.send(buyerReviewCreatedTopic, BuyerReviewCreatedMessage.from(event));

        future.addCallback(new ListenableFutureCallback<>() {
            @Override
            public void onFailure(Throwable ex) {
                log.warn("handleBuyerReviewCreatedEvent exception occurred with sellerId: {}, reviewerId: {}: {}",
                        sellerId, reviewerId, ex.getMessage(), ex);
            }

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

---

Consumer 예외 핸들링 적용

• DefaultErrorHandler 빈을 공통으로 정의하고 해당 값에 FixedBackOff값을 설정
  • FixedBackOff 설정은 1초 간격으로 최대 3회 재실행을 시도한다.
• 지금 단계에서는 별도의 Dead Message Topic을 정의하기보다는 몇 번의 재시도 후 로깅 남기는 것으로 처리
• DefaultErrorHandler을 공통으로 정의하고 모든 리스너에서 .setCommonErrorHandler() 메서드를 사용해서 주입받음

@Slf4j
@Configuration
public class KafkaListenerConfig {

    //...

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, SoldOutMessage> kafkaListenerContainerSoldOutFactory() {
        //...
        factory.setCommonErrorHandler(defaultErrorHandler());
        return factory;
    }

    //...

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, PriceDownMessage> kafkaListenerContainerPriceDownFactory() {
        //...
        factory.setCommonErrorHandler(defaultErrorHandler());
        return factory;
    }

    //...

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, SoldOutToBuyerMessage> kafkaListenerContainerSoldOutToBuyerFactory() {
        //...
        factory.setCommonErrorHandler(defaultErrorHandler());
        return factory;
    }

    //...

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, BuyerReviewCreatedMessage> kafkaListenerContainerBuyerReviewCreatedFactory() {
        //...
        factory.setCommonErrorHandler(defaultErrorHandler());
        return factory;
    }

    //...

    @Bean
    public DefaultErrorHandler defaultErrorHandler() {
        return new DefaultErrorHandler(((consumerRecord, e) ->
            log.warn("defaultErrorHandler invoked with consumerRecord: {}, message: {}",
            consumerRecord, e.getMessage(), e)), new FixedBackOff(1000L,3L));
    }
}