kafka

Spring kafka의 concurrency는 높으면 좋은걸까?

Spring Kafka를 사용할 때 아래와 같이 concucrency라는 옵션이 있다. 이 값이 높으면 성능이 좋아질까? @KafkaListener( topics = "kafka-concurrency", groupId = "kafka-concurrency-consumer-group", concurrency = "1" ) 테스트 방법 concurrency의 동작을 확인하기 위해 아래와 같은 시나리오로 테스트하였다. producer는 10개의 메시지를 동시에 보낸다. for (int i = 0; i < 10; i++) { TextMessage message = new TextMessage("message " + i); log.info("[send] " + message.getText()); kafkaTemp..

2023. 1. 20. 09:46

kafka

KafkaListener에서 서로 다른 message mapping하기

다른 2개의 서비스에서 동일 토픽을 다른 클래스(서비스내 클래스)로 매핑하는 방법이다. [A서비스에서 메시지 발행] topicId: message-topic message는 2가지 발행: service1.model.TextMessage, service1.model.FileMessage [B서비스에서 메시지 구독] topicId: message-topic message는 2가지 발행: service2.model.TextMessage, service2.model.FileMessage spring kafka에서 발행한 메시지의 헤더를 보면 __TypeID__라는 것이 있다. 이것은 메시지를 보낼 때의 데이터 객체 타입을 매핑하여 전송하게 되고 실제 수신받는 컨슈머는 이 객체타입을 사용하여 deserialize..

2022. 12. 23. 07:36

kafka

kafka publisher의 성능 비교

[궁금점] kafka publisher로 메시지 전송 시 아래와 같은 궁금점에 대해 테스트해 보았다. 동기(sync)와 비동기(async)의 성능 차이가 얼마나 날까? topic의 partition 수를 늘리면 성능이 빨라질까? producer의 acks 설정이 성능에 어떤 영향을 줄까? [테스트 환경] 개발환경: spring boot, java 8, spring-kafka kafka는 원격서버에 위치 single-kafka: brokder 1대 multi-kafka: broker 3대 (replication-factor: 3) [테스트 방법] 메시지 내용: 100개의 문자열 publisher를 1000번 실행한 시간에 대한 3회 평균 시간 [테스트 코드] // 동기 @Transactional publi..

2022. 12. 23. 07:13

kafka

kafka CommitFailedException 정리

Kafka에서 CommitFailedException이 나오는 유형은 아래와 같다. consumer 로직의 처리시간이 max.poll.interval.ms보다 클 경우 리밸런싱으로 인한 컨슈머 그룹에서 제외되었을 경우 session.timeout.ms시간동안 heartbeat가 오지 않았을 경우 리밸런싱으로 인한 컨슈머 그룹에서 제외되었을 경우 1. consumer 로직의 처리시간이 max.poll.interval.ms보다 클 경우 리밸런싱으로 인한 컨슈머 그룹에서 제외되었을 경우 ConsumerConfig에서 @Bean public ConsumerFactory consumerFactory() { ... props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 50..

2022. 12. 23. 07:13