카프카를 프로그램으로 관리하기 위한 API. 토픽 목록 조회와 생성, 삭제, 클러스터 상세 정보 확인, ACL 관리, 설정 변경을 할 수 있다.
토픽이 없을 때의 대처
“이벤트를 써야 하는데 토픽이 없다”는 상황을 처리할 방법이 마땅치 않았다. producer.send()에서 UNKNOWN_TOPIC_OR_PARTITION 예외를 잡아 사용자에게 토픽을 만들라고 알려주거나, 자동 토픽 생성 기능을 켜두거나, 호환성을 포기하고 내부 API를 쓰는 정도였다. 어느 쪽도 사용자 친화적이지 않다. AdminClient가 있으면 토픽이 있는지 확인하고 없으면 즉석에서 만든다.
비동기 결과와 에러 처리
모든 메서드가 Future 객체를 리턴한다. 정확히는 Result 객체가 Future를 감싼 형태다.
ListTopicsResult topics = admin.listTopics();
topics.names().get().forEach(System.out::println);get()은 블로킹이다. 어드민 작업은 드물어서 대개 블로킹으로 써도 되지만, 많은 어드민 요청을 처리하는 서버라면 whenComplete()로 비동기 처리해서 스레드를 놀리지 않는다.
카프카가 에러 응답을 보내면 ExecutionException이 발생한다. Future 객체 안에 예외가 다시 들어 있는 구조라서, 카프카가 리턴한 진짜 에러를 보려면 항상 ExecutionException의 cause를 확인해야 한다.
} catch (ExecutionException e) {
if (!(e.getCause() instanceof UnknownTopicOrPartitionException)) throw e;
// 여기까지 왔으면 토픽이 없는 것
}목록을 다루는 메서드는 에러 처리 방식을 고를 수 있다. valid()는 클러스터가 에러 없이 리턴한 것만 주고, errors()는 발생한 모든 예외를 주고, all()은 첫 번째 에러만 예외로 던진다.
평평한 인터페이스
인터페이스가 평평하다는 것도 알아둘 만하다. 카프카 프로토콜이 단순 네트워크 메시지 수준에서 동작하므로 객체 간 의존 관계나 네임스페이스가 없다. 토픽, 브로커, 파티션은 문자열이나 숫자 ID로 식별되는 엔티티일 뿐이고 토픽 이름이 곧 유일 식별자다. 그래서 AdminClient에 어떤 기능이 없다면 그것은 정말로 구현되지 않은 것이다. 인터페이스는 크지만 IDE 자동완성으로 찾기 편하다.
설정과 오프셋, 레코드 다루기
설정 관리는 ConfigResource로 브로커, 브로커 로그, 토픽 설정을 다룬다. 브로커 설정은 kafka-configs.sh로 하는 것이 보통이고, 토픽 설정 확인과 수정은 애플리케이션에서 아주 흔하다. 정확한 작동을 위해 압착이 걸려 있어야 하는 토픽이라면 보존 기한보다 짧은 주기로 확인하고 아니면 고치는 코드를 둔다. 변경은 SET, DELETE, APPEND, SUBTRACT 네 가지 작업 유형으로 한다.
컨슈머 그룹 오프셋 변경은 가장 유용하게 쓰인다. 이슈가 생긴 애플리케이션을 트러블슈팅하거나 재해 복구 상황에서 새 클러스터로 옮길 때 오래된 메시지를 다시 읽어야 하기 때문이다.
레코드 삭제는 deleteRecords()로 한다. 호출 시점 기준으로 지정된 오프셋보다 오래된 레코드에 삭제 표시를 해서 컨슈머가 접근하지 못하게 하고, 디스크에서 실제로 지우는 작업은 비동기적으로 일어난다. 세그먼트가 안 닫혀서 보존 기한을 넘겨도 안 지워지는 경우에 쓴다.
레플리카 재할당은 alterPartitionReassignments()로 각 레플리카의 위치를 정밀하게 제어한다. 대량의 데이터 복제가 일어나므로 사용 가능한 네트워크 대역폭에 주의하고, 필요하면 쿼터를 설정해 복제 작업을 스로틀링한다.
테스트에는 MockAdminClient가 있다. 원하는 수의 브로커를 설정해 초기화할 수 있어서 실제 클러스터를 띄울 필요가 없다.
최종적 일관성
컨트롤러에서 브로커로의 메타데이터 전파가 비동기적이라서, 리턴된 Future는 컨트롤러의 상태가 업데이트된 시점에 완료된 것으로 간주된다. 이것이 최종적 일관성(eventual consistency)이다. 최종적으로 모든 브로커가 알게 되지만 그 시점이 언제인지는 보장하지 않으므로, createTopics() 직후에 describeTopics()를 부르면 일관되지 않은 결과가 나올 수 있다.
오프셋과 파티션 변경의 제약
작동 중인 컨슈머 그룹의 오프셋은 수정할 수 없다. 시도하면 UnknownMemberIdException이 나는데, 코디네이터 입장에서는 그룹 멤버가 아닌 클라이언트가 오프셋을 커밋하려는 것으로 보이기 때문이다. 반드시 컨슈머 애플리케이션을 정지시키고 변경해야 하며, 특정 컨슈머 그룹만 정지시키는 어드민 명령은 없다.
오프셋을 삭제하는 것과 명시적으로 맨 앞으로 옮기는 것도 다르다. 삭제하면 컨슈머가 어디서 시작할지는 auto.offset.reset 설정에 달리게 되어 예측할 수 없다. 명시적으로 맨 앞으로 옮겨야 확실히 처음부터 읽는다.
상태를 가진 애플리케이션은 오프셋만 되돌리면 안 된다. 상점에서 팔린 신발 수를 집계하는 스트림 애플리케이션에서 오전 8시에 문제를 발견하고 오전 3시로 오프셋만 돌리면, 저장된 집계값 위에 다시 더해져서 중복 계산이 된다. 상태 저장소도 함께 비워야 한다.
파티션 추가는 드물고 위험하다. createPartitions()를 쓰는데 키 기반 파티셔닝이 깨지기 때문이고, 추가할 개수가 아니라 추가된 뒤의 총 개수를 지정한다는 점도 함정이다. 토픽을 읽고 있는 애플리케이션들이 깨지지 않는지 확인하고 써야 한다.
MockAdminClient는 카프카 API의 일부가 아니어서 언제든 바뀔 수 있다. 다만 공개된 메서드에 대한 목업이라 시그니처는 호환성을 유지한다.