OPEN

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

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 속성