프로듀서

• 토픽에 메시지 전송
  • 토픽, 키, 값

Properties prop = new Properties();
prop.put("bootstrap.servers", "kafka01:9092,kafka01:9092,kafka01:9092");
prop.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
prop.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 
KafkaProducer<Integer, String> producer = new KafkaProducer<>(prop);
 
producer.send(new ProducerRecord<>("topicname", "key", "value"));
producer.send(new ProducerRecord<>("topicname", "value"));
 
producer.close();

producer.send(new ProducerRecord<>("simple", "value"));

• 전송 실패를 알 수 없음
• 실패에 대한 별도 처리가 필요없는 메시지 전송에 사용

Future<RecordMetadata> f = producer.send(new ProducerRecord<>("topic", "value"));

try {
  RecordMetadata meta = f.get(); // 블로킹
} catch (ExecutionException ex) {

}

• 배치 효과 떨어짐 → 처리량 저하
• 처리량이 낮아도 되는 경우에만 사용

producer.send(new ProducerRecord<>("simple", "value"),
  new Callback() {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception ex) {
    }
  });

• 처리량 저하 없음

• ack = 0
  • 서버 응답을 기다리지 않음
  • 전송 보장도 zero
• ack = 1
  • 파티션의 리더에 저장되면 응답 받음
  • 리더 장애시 메시지 유실 가능
• ack = all (또는 -1)
  • 모든 리플리카에 저장되면 응답 받음
    • 브로커 min.insync.replicas 설정에 따라 달라짐

• min.insync.replicas (브로커 옵션)
  • 프로듀서 ack 옵션이 all일 때 저장에 성공했다고 응답할 수 있는 동기화된 리플리카 최소 개수
• 예1:
  • 리플리카 개수 3, ack = all, min.insync.replicas = 2
  • 리더에 저장하고 팔로워 중 한 개에 저장하면 성공 응답
• 예2:
  • 리플리카 개수 3, ack = all, min.insync.replicas = 1
  • 리더에 저장되면 성공 응답
  • ack = 1 과 동일 (리더 장애시 메시지 유실 가능)
• 예3:
  • 리플리카 개수 3, ack = all, min.insync.replicas = 3
  • 리더와 팔로워 2개에 저장되면 성공 응답
  • 팔로워 중 한 개라도 장애가 나면 리플리카 부족으로 저장에 실패함

• 전송 과정에서 실패
  • 전송 타임 아웃 (일시적인 네트워크 오류 등)
  • 리더 다운에 의한 새 리더 선출 진행 중
  • 브로커 설정 메시지 크기 한도 초과
  • 등등
• 전송 전에 실패
  • 직렬화 실패, 프로듀서 자체 요청 크기 제한 초과
  • 프로듀서 버퍼가 차서 기다린 시간이 최대 대기 시간 초과
  • 등등

• 재시도
  • 재시도 가능한 에러는 재시도 처리
    • 예: 브로커 응답 타임 아웃, 일시적인 리더 없음 등
• 재시도 위치
  • 프로듀서는 자체적으로 브로커 전송 과정에서 에러가 발생하면 재시도 가능한 에러에 대해 재전송 시도
    • retries 속성
  • send() 메서드에서 익셉션 발생시 익셉션 타입에 따라 send() 재호출
  • 콜백 메서드에서 익셉션 받으면 타입에 따라 send() 재호출
• 아주 아주 특별한 이유가 없다면 무한 재시도 ❌

• 추후 처리 위해 기록
  • 별도 파일, DB 등을 이용해서 실패한 메시지 기록
  • 추후에 수동(또는 자동) 보정 작업 진행
• 기록 위치
  • send() 메서드에서 익셉션 발생시
  • send() 메서드에 전달한 콜백에서 익셉션 받은 경우
  • send() 메서드가 리턴한 Future의 get() 메서드에서 익셉션 발생시

• 브로커 응답이 늦게 와서 재시도할 경우 중복 발송 가능
• 참고 : enable.idempotence 속성

• max.in.flight.requests.per.connection
  • 블록킹없이 한 커넥션에서 전송할 수 있는 최대 전송중인 요청 개수
  • 이 값이 1보다 크면 재시도 시점에 따라 메시지 순서가 바뀔 수 있음
    • 전송 순서가 중요하면 이 값을 1로 지정

• https://www.youtube.com/watch?v=geMtm17ofPY