Kafka

왜 프로듀서 설계가 유스케이스마다 달라지는가

Kafka에 메시지를 쓰는 이유는 사용자 활동 기록, 메트릭 수집, 로그 저장, 애플리케이션 간 비동기 통신, DB 쓰기 전 버퍼링 등 매우 다양하다. 유스케이스가 다르면 요구사항도 달라진다. 모든 메시지가 critical한가, 아니면 일부 유실을 감내할 수 있는가? 중복이 생겨도 괜찮은가? 지연이나 처리량에 엄격한 제약이 있는가?

책이 드는 예로, 신용카드 거래 처리 시스템은 단 한 건도 유실하거나 중복해서는 안 되고, 지연은 낮아야 하지만 500ms까지는 감내할 수 있으며, 초당 100만 건까지 처리해야 한다. 반대로 웹사이트 클릭 정보 수집은 약간의 유실이나 중복이 허용되고, 사용자 경험에 영향이 없다면 메시지가 Kafka에 몇 초 늦게 도착해도 괜찮다.

이 요구사항 차이가 곧 프로듀서 API를 쓰는 방식과 설정 선택으로 이어진다. 프로듀서 설정을 외우는 것보다, 내 유스케이스가 유실·중복·지연·처리량 네 축에서 어디에 서 있는지를 먼저 정하는 것이 순서다.

ProducerRecord가 브로커에 도달하기까지

프로듀서로 메시지를 보내는 과정은 ProducerRecord 객체를 만드는 것으로 시작한다. ProducerRecord는 보낼 토픽과 값(value)을 반드시 포함해야 하고, 키(key)와 파티션 번호는 선택적으로 지정할 수 있다.

send()를 호출하면 프로듀서 내부에서 다음 순서로 처리된다.

1. 직렬화: 키와 값 객체를 네트워크로 보낼 수 있도록 바이트 배열로 직렬화한다. 2. 파티셔너: 직렬화된 데이터가 파티셔너로 넘어간다. ProducerRecord에 파티션을 명시했다면 파티셔너는 아무 일도 하지 않고 그 파티션을 그대로 반환한다. 명시하지 않았다면 보통 키를 기준으로 파티션을 골라 준다. 3. 배치 적재: 토픽과 파티션이 정해지면, 같은 토픽·파티션으로 갈 레코드들의 배치에 이 레코드를 추가한다. 4. 전송: 별도의 스레드가 이 배치들을 해당 Kafka 브로커로 전송한다.

브로커가 메시지를 받으면 응답을 돌려준다. 성공하면 토픽·파티션·파티션 내 오프셋을 담은 RecordMetadata 객체를 반환하고, 실패하면 오류를 반환한다. 프로듀서가 오류를 받으면 몇 차례 재시도한 뒤에도 실패하면 그때 오류를 애플리케이션에 알린다.

여기서 중요한 것은 send()가 곧바로 네트워크 전송을 일으키지 않는다는 점이다. 레코드는 일단 버퍼에 들어가고 별도 스레드가 나중에 보낸다. 이 비동기 구조가 뒤에서 볼 배치·압축·순서 문제의 근원이다.

프로듀서 생성에 필수인 세 가지 설정

KafkaProducer를 만들려면 Properties에 반드시 세 가지를 넣어야 한다.

bootstrap.servers는 프로듀서가 클러스터에 최초 연결할 때 쓰는 브로커의 host:port 목록이다. 모든 브로커를 나열할 필요는 없다. 최초 연결 후 프로듀서가 나머지 정보를 받아 오기 때문이다. 다만 브로커 한 대가 죽어도 연결할 수 있도록 최소 두 개는 넣는 것이 권장된다.

key.serializer는 레코드 키를 바이트 배열로 바꿀 클래스 이름이다. Kafka 브로커는 키와 값을 바이트 배열로만 다루지만, 프로듀서 인터페이스는 제네릭 타입으로 임의의 Java 객체를 받을 수 있게 해 준다. 그래서 그 객체를 바이트 배열로 바꾸는 방법을 프로듀서에게 알려 줘야 한다. org.apache.kafka.common.serialization.Serializer 인터페이스를 구현한 클래스여야 하며, Kafka 클라이언트 패키지에 ByteArraySerializer, StringSerializer, IntegerSerializer가 이미 들어 있다. 주의할 점은 값만 보낼 생각이더라도 key.serializer는 반드시 설정해야 한다는 것이다.

value.serializer는 값 객체를 바이트 배열로 바꿀 클래스다. 역할은 key.serializer와 동일하다.

private Properties kafkaProps = new Properties();
kafkaProps.put("bootstrap.servers", "broker1:9092,broker2:9092");
kafkaProps.put("key.serializer",
  "org.apache.kafka.common.serialization.StringSerializer");
kafkaProps.put("value.serializer",
  "org.apache.kafka.common.serialization.StringSerializer");
producer = new KafkaProducer<String, String>(kafkaProps);

인터페이스가 이렇게 단순하다는 것은 곧 프로듀서 동작의 대부분을 설정 프로퍼티로 제어한다는 뜻이다.

스레딩 측면에서, 이 장의 예제는 모두 단일 스레드지만 프로듀서 객체 자체는 여러 스레드가 공유해 메시지를 보내도 안전하다. 처음에는 프로듀서 하나에 스레드 하나로 시작하고, 처리량이 부족하면 같은 프로듀서를 쓰는 스레드를 늘린다. 그래도 더 늘지 않으면 그때 프로듀서 인스턴스를 추가한다.

전송 방식 1 — Fire-and-forget

가장 단순한 전송은 send()를 호출하고 반환값을 무시하는 것이다.

ProducerRecord<String, String> record =
  new ProducerRecord<>("CustomerCountry", "Precision Products", "France");
try {
  producer.send(record);
} catch (Exception e) {
  e.printStackTrace();
}

send()는 RecordMetadata를 담은 Java Future를 반환하지만, 이 값을 무시하면 메시지가 성공적으로 전송됐는지 알 방법이 없다. Kafka는 가용성이 높고 프로듀서가 자동 재시도를 하므로 대부분은 잘 도착하지만, 일부 메시지는 이 방식에서 유실된다. 메시지를 조용히 버려도 괜찮은 경우에만 쓸 수 있으며, 프로덕션에서는 보통 해당하지 않는다.

주의할 점은 이 코드의 try/catch가 잡는 예외의 범위다. 브로커에서 발생한 오류나 전송 중 오류는 잡히지 않는다. 잡히는 것은 프로듀서가 브로커로 보내기 전에 발생한 오류뿐이다. 직렬화 실패로 인한 SerializationException, 버퍼가 가득 찼을 때의 BufferExhaustedException 또는 TimeoutException, 전송 스레드가 인터럽트됐을 때의 InterruptException이 여기에 해당한다.

전송 방식 2 — 동기 전송

동기 전송은 send()가 반환한 Future에 대해 get()을 호출해 응답을 기다리는 방식이다.

ProducerRecord<String, String> record =
  new ProducerRecord<>("CustomerCountry", "Precision Products", "France");
try {
  producer.send(record).get();
} catch (Exception e) {
  e.printStackTrace();
}

Future.get()은 전송이 실패하면 예외를 던지고, 성공하면 RecordMetadata를 돌려주므로 메시지가 기록된 오프셋을 알 수 있다. 예외가 발생하는 경우는 (1) Kafka로 데이터를 보내기 전 오류, (2) 전송 중 오류, (3) 브로커가 재시도 불가능한 예외를 반환한 경우, (4) 사용 가능한 재시도 횟수를 모두 소진한 경우다.

동기 전송은 정확하지만 느리다. 애플리케이션과 클러스터 사이 네트워크 왕복이 10ms라면, 매번 응답을 기다리면서 메시지 100건을 보내는 데 약 1초가 걸린다.

전송 방식 3 — 비동기 전송과 콜백

응답을 아예 기다리지 않으면 100건을 보내는 데 시간이 거의 들지 않는다. 사실 대부분의 경우 응답 자체는 필요 없다. Kafka가 돌려주는 토픽·파티션·오프셋은 보내는 쪽에서 쓸 일이 별로 없기 때문이다. 하지만 전송에 완전히 실패했다는 사실은 알아야 한다. 예외를 던지거나, 로그를 남기거나, 나중에 분석할 수 있게 오류 파일에 기록해야 하기 때문이다.

그래서 프로듀서는 레코드를 보낼 때 콜백을 함께 등록할 수 있게 해 준다.

private class DemoProducerCallback implements Callback {
    @Override
    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
        if (e != null) {
            e.printStackTrace();
        }
    }
}

ProducerRecord<String, String> record =
  new ProducerRecord<>("CustomerCountry", "Biomedical Materials", "USA");
producer.send(record, new DemoProducerCallback());

콜백을 쓰려면 org.apache.kafka.clients.producer.Callback 인터페이스를 구현해야 한다. 이 인터페이스에는 onCompletion() 하나의 메서드만 있다. Kafka가 오류를 반환했다면 onCompletion()의 Exception 인자가 null이 아니다. 예제에서는 출력만 하지만 실제 코드에서는 더 견고한 오류 처리를 넣어야 한다.

비동기 방식에서 지연은 감춰지지만, 처리량의 상한은 in-flight 메시지 수, 즉 응답을 받기 전에 프로듀서가 보낼 수 있는 메시지 개수로 결정된다.

재시도 가능한 오류와 재시도 불가능한 오류

KafkaProducer가 만나는 오류는 두 종류다.

재시도 가능한(retriable) 오류는 메시지를 다시 보내면 해결될 수 있는 오류다. 연결 오류는 연결이 다시 맺어지면 해결되고, "no leader" 오류는 해당 파티션에 새 리더가 선출되면 해결된다. KafkaProducer는 이런 오류를 자동으로 재시도하도록 설정할 수 있으며, 그렇게 하면 애플리케이션 코드는 재시도를 모두 소진하고도 해결되지 않은 경우에만 재시도 가능 예외를 보게 된다.

재시도로 해결되지 않는 오류도 있다. "message size too large"가 대표적이다. 이런 경우 KafkaProducer는 재시도를 아예 시도하지 않고 즉시 예외를 반환한다.

이 구분은 애플리케이션 코드의 역할을 정해 준다. 프로듀서가 재시도를 대신해 주므로 애플리케이션 로직에 재시도를 직접 넣을 이유가 없다. 대신 재시도 불가능한 오류와 재시도를 모두 소진한 경우를 처리하는 데 집중해야 한다.