idempotence(멱등성) 기반 중복 없이 전송

최대 한번 전송/적어도 한번 전송/ 정확히 한번 전송

• 최대 한번 전송(at most once): 중복 불허, acks = 0
• 적어도 한번 전송(at least once): 중복 허용, retry > 0, acks=1, all
• 정확히 한번 전송(exactly once)
  • 중복 없이 전송(Idempotence): Producer의 message 전송! retry 시, 중복 제거
  • Transaction 기반 전송: Consumer > Process > Producer(주로 Kafka Streams)에 주로 사용되는 Transaction 기반 처리

최대 한번 전송

acks = 0

• Producer는 브로커로부터 Ack 또는 에러 메시지 없이 다음 메시지를 연속으로 보낸다.
• 메시지가 소실될 수 있지만 중복 전송은 하지 않는다.

적어도 한번 전송

acks = 1이상, all

• Producer는 브로커로부터 Ack를 받은 다음에 다음 메시지를 전송
• 메시지 소실은 없지만 중복 전송을 할 수 있음

중복 없이 전송

acks = all, enable.indempotence = true, retries > 0, 1 < max.inflight.requests.per.connection < 5

• 동작 과정
  • 메시지 A가 정상적으로 브로커에 기록되고, Ack 전송(메시지  A는 프로듀서 ID:0, Seq: 0으로 브로커 메모리에 저장)
  • producer는 Ack를 기다린 후 메시지 B를 전송
  • 메시지 B가 정상적으로 브로커에 기록되었지만, 네트워크 장애로, Ack를 Producer에 보내지 못함(메시지 B는 Seq:1로 브로커에 저장)
  • 메시지 B의 Ack를 받지 못한 Producer가 메시지 B를 다시 보낸다.
  • 메시지  B는 재전송 되었지만 브로커는 Seq가 1인 메시지가 이미 저장되었기 때문에 해당 메시지를 메시지 로그에 저장하지 않고, Ack만 응답한다.
• Producer는 브로커로부터 Ack를 받은 다음에 다음 메시지 전송을 하되, Producer Id, 메시지 Seq를 Header에 저장하여 전송
• 메시지 Seq는 메시지 고유의 Seq 번호로, 0부터 시작하여 순차적으로 증가 -> ProducerId는 producer 기동시마다 새로 생성
• 브로커에서 메시지 Seq가 중복될 경우 메시지 로그에 기록하지 않고 Ack만 전송
• 브로커는 Producer가 보낸 메시지의 Seq가 브로커가 가지고 있는 메시지의 Seq보다 1만큼 큰 경우에만 브로커에 저장

Idempotence 적용 후에는 성능이 약간 감소(최대 20%) 할 수 있지만 기본적으로 idempotence 적용을 권장

Idempotence 설정 시, 유의사항

• Kafka 3.0 버전부터는 Producer의 기본 설정이 Idempotence이다.
• 기본 설정 주 enable.idempotence=true를 제외하고 다른 파라미터를 잘못 설정하면, Producer는 정상적으로 메시지를 보내지만 idempotence로 동작하지 않음
• 명시적으로 enable.idempotence=true 를 설정한 뒤, 다른 파라미터를 잘못 설정하면, Config  오류가 발생하면서 Producer가 기동되지 않음

Idempotence 기반에서 메시지 전송 순서 유지

• B0 > B1 >  B2 순에서 Producer에서 생성된 메시지 배치
• Idempotence 기반에서 max.in.flight.requests.per.connection 만큼 여러개의 배치들이 브로커에 전송된다.
• 브로커는 메시지 배치를 처리 시, write된 배치의 마지막 메시지 seq + 1이 아닌 배치 메시지가 올 경우에는 OutofOrderSequenceException을 생성하여, Producer에 오류로 전달

Idempotence 정리

• Idempotence는 Producer와 브로커간에 메시지 retry시에만 중복 제거를 수행하는 메커니즘
• 동일한 메시지를 send()로 두번 호출하여 전송하는 것은 Idempotence의 중복 제거 대상이 아니다.
• Producer가 재기동되면, ProducerID가 달라지므로 이전 Producer가 전송한 메시지를 중복해서 보낼 수 있다.
• Consumer > Process > Producer 로직에서 Consumer가 _consumer_offsets에 읽은 offset을 저장하지 못하였지만 , Producer는 메시지를 전송하였다면, Consumer는 이전 offset을 읽어서 producer를 통해 중복된 메시지를 전송할 수 있다.(트랜젝션 기반 처리 필요)

 

실습하기

1) 기본 enable.idempotence 설정은 true이다(디폴트)

기본값

2) 다른 Property를 변경하게 되면, Idempotence 기본값 변경

Properties props = new Properties();

props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.56.101:9092");
props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.setProperty(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "6");
props.setProperty(ACKS_CONFIG, "0");

• Idempotence로 가질 않는다.(확인할 방법이 없음..)
  • 문서 확인
  • https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html

Kafka Producer Configuration Reference for Confluent Platform | Confluent Documentation

 

3) 명시적으로 Idempotence를 입력하면 오류 발생

Properties props = new Properties();

props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.56.101:9092");
props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.setProperty(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "6");
props.setProperty(ACKS_CONFIG, "0");
props.setProperty(ENABLE_IDEMPOTENCE_CONFIG, "true");

• 명시적으로 입력하게 되면, 실행시 오류가 발생한다.