토픽 목록 조회, 생성, 삭제, 클러스터 상세 정보 등의 프로그램적 관리를 위해 제공되는 API를 말한다.
명령줄로도 특정 정보는 확인이 가능하지만, 사용자 입력에 기반한 관리를 위해서는 해당 API가 필요하다.
특징
비동기
클러스터 컨트롤러 API 요청을 보내면, 1개 이상의 Future 객체를 리턴한다.
Future는 비동기적 연산 결과를 표현하는 인터페이스이다.
이 객체들은 컨트롤러의 상태가 완전히 업데이트된 시점에서 완료된 것으로 간주한다.
Future 객체를 감싼 Result 객체를 반환한다. Result 객체는 작업이 끝날때까지 대기하거나 적업 결과에 대해 일반적으로 뒤이어 쓰이는 작업을 수행하는 헬퍼 메서드를 가지고 있다.
public class CreateTopicsResult {
private final Map<String, KafkaFuture<TopicMetadataAndConfig>> futures;
protected CreateTopicsResult(Map<String, KafkaFuture<TopicMetadataAndConfig>> futures) {
this.futures = futures;
}
...
}
최종적 일관성 (eventual consistency)
분산시스템의 일관성을 나타내는 정도로, 시스템의 모든 복제본이 결국 최종적으로 같은 값을 가지게 되는 모델을 뜻합니다.
해당 뜻은, 최종적으로 같은 값을 가지기 까지 시간이 얼마나 걸리는지는 알 수 없음을 의미하고도 한다.
일시적인 불일치를 허용하고, 가용성에 집중한 일관성 모델이다.
옵션 (Options)
AdminClient 메서드들은 각 메서드에 맞는 Options 객체를 인수로 받는다.
해당 객체는 요청 받은 브로커가 요청을 어떻게 처리할지에 대한 설정을 담고 있다.
공통적으로 모든 Options 객체는 timeoutMs 메서드를 가지고 있으며, 클러스터로부터의 응답을 기다리는 시간을 설정한다.
그외에도, 각각 Options 객체마다 요청을 어떻게 처리할지에 대한 설정을 할 수 있는 메서드들이 있다. ( `ListTopicsOptions`는 `listInternal(boolean)`로 내부 토픽을 리턴할지 결정하고, `DescribeClusterOptions`는 `includeAuthorizedOperations(boolean)`로 권한을 가진 직업을 리턴할지를 결정한다.)
// Options 객체
package org.apache.kafka.clients.admin;
public abstract class AbstractOptions<T extends AbstractOptions> {
protected Integer timeoutMs = null;
public AbstractOptions() {
}
public T timeoutMs(Integer timeoutMs) {
this.timeoutMs = timeoutMs;
return this;
}
public Integer timeoutMs() {
return this.timeoutMs;
}
}
수평구조
어떠한 의존관계나 네임스페이스를 가지지 않는다. 실제로 코드를 살펴보면 하나의 `KafkaAdminClient`객체 안에 모든 객체 및 메서드가 구현되어 있다.
💡 어드민 클라이언트가 클러스터에게 create, delete, alter 요청을 보내게되면 해당 작업은 컨트롤러에 의해 수행된다. 반면, 단순 일기만 하는 작업인 list, describe는 부하가 작은 브로커에게 전달되게 된다. 그래서 해당 브로커가 최신의 변경사항(예를 들어 새로운 토픽이 추가되는 등)을 알고 있지 않다면 최신 내용은 반환하지 않는다는 점을 유의하자.
이것이 최종적 일관성 속성을 가지는 이유이기도 하다.
💡 어드민 작업은 크게 AdminClient를 통해 진행하거나, 주키퍼에 저장되어 있는 메타데이터를 직접 수정하는 방식으로 진행할 수 있다. 하지만 책은 주키퍼로 진행하는 방식은 사용하지 말라고 강력하게 말하고 있다. 이유는 아래와 같다.
1. 메타데이터를 아파치 카프카와 주키퍼에 저장하게 되면서 중복의 문제 및 데이터 일관성의 문제가 있다. 2. 외부에서 메타데이터를 관리하는 것은 효율성이 떨어진다. 3. 카프카의 확장성이 제한된다.
Apache ZooKeeper의 문제가 아닌 외부 메타데이터 관리 방식의 문제점임을 유의하자.
브로커 중 하나에 장애가 발생할 경우에 대비해서 최소 3개 이상의 브로커를 지정하는 것이 보통이다.
어드민 클라이언트를 사용완료했다면, `close()`를 통해 닫아줘야 한다.
시간을 입력을 받는데, 해당 시간만큼 응답을 기다리는 것을 의미한다.
close()가 실행된 시점에 아직 진행중인 작업이 있을 수 있는데,해당 작업을 위해 응답을 기다리는 시간을 가지는 것이다.
만약 타임아웃이 된다면, 모든 작동을 멈추고 자원을 해제한다.
시간 값을 넣어주지 않는다면, 모든 진행 중이 작업이 완료될때 까지 기다린다는 것을 의미한다.
설정
1. client.dns.lookup ( 기본값 : default )
기본적으로 카프카는 부트스트랩 서버 설정에 포함된 호스트명을 기준으로 연결을 시도한다. (default)
웬만한 경우 정상적으로 동작하지만, DNS를 사용하는 상황에서는 다른 방법이 필요했다.
그래서 해당 설정값이 등장했고, DNS를 사용하는 상황도 2가지로 나뉘기에 각 상황에 따른 설정하는 방벙에 대해 알아보자.
DNS 별칭을 사용하는 경우
다양한 브로커가 존재하는 상황에서 모든 브로커를 부트스트랩 설정에 일일이 지정하는 것은 유지관리에 취약하다. 그때 하나의 DNS 별칭으로 전체의 브로커는 가리킬 수 있다.
그런데, 해당 경우 `SASL`을 사용하면 인증을 하려고 할때 문제 발생한다.
부트스트랩에 설정되어 있는 DNS 별칭으로 인증을 시도하려고 하는데, SASL은 서버의 보안 주체와 호스트명이 일치하지 않으면 인증을 거부하게된다.
그때 `client.dns.lookup = resolve_cannonical_bootstrap_servers_only`로 설정해주자.
해당 설정은 DNS 별칭을 펼치게 도아주면, 모든 브로커의 이름을 일일이 부트스트램 서버 목록에 넣어주는 것과 동일한 역할을 수행한다.
💡 카프카 인증 : SSL 과 SASL
카프카에서 인증을 처리하는 방식은 크게 SSL과 SASL이 있다.
SSL은 대칭키와 공개키를 활용한 암호화 통신 방식으로, 클라이언트에게 메세지가 암호화 되어 전달된다. 즉, 해당 메세지를 복호화하는 작업이 필요로 하고, 그만큰 CPU 부하가 발생한다.
SASL은 프레임워크로 인증 처리 방식을 plug-in 형태로 요구사항에 맞게 변경하여 사용하는 것이 특징이다. 문자열 아이디, 패스워드를 사용하는 인증(PLAIN), SCRAM 방식을 이용한 인증, OAuth2 방식을 이용한 인증(OAUTHBEARER), Kerberos를 이용한 인증(GSSAPI)이 있다.
다수의 IP주소로 연결되는 DNS 이름을 사용하는 경우
로드 밸런서 뒤에 모든 브로커를 숨기는 경우, 로드 밸런서가 단일 장애점이 되는 것을 방지하기 위해 브로커를 여러개의 IP 주소로 연결한다.
카프카 클라리언트는 첫번째 호스트명으로 연결시도하게 되고, 만약 해석된 IP가 불능 상태라면 다른 IP가 정상적으로 동작하더라도 연결 실패를 띄운다.
public static void main(String[] args) throws ExecutionException, InterruptedException {
Map<String, NewPartitions> newPartitions = new HashMap<>();
newPartitions.put(TOPIC_NAME, NewPartitions.increaseTo(NUM_PARTITIONS+2));
adminClient.createPartitions(newPartitions).all().get();
}
public class NewPartitions {
private int totalCount; // 파티션 수
private List<List<Integer>> newAssignments; // 새로운 파티션에 대한 복제 할당
}
`NewPartitions.increaseTo()`를 사용하면 파티션을 추가할 수 있다.
NewPartitions 객체는 파티션의 수인 totalCount 와 레플리카 할당에 관한 newAssignments를 가지고 있다.
increaseTo()는 해당 객체를 생성하는 생성자 역할을 하며, 기존 파티션수 + 추가할 파티션 수를 넣어주면 된다.
createPartitions() 사용해 최종적으로 지정된 토픽에 파티션을 추가한다.
여러 토픽을 한번에 확장할 경우 일부는 성공하고, 나머지는 실패할 수 도 있으니 주의하자.
위에서 봤듯이 해당 방법을 사용하기 위해서는 토픽 상세 정보를 통해 기존의 파티션의 수를 알아야 한다.
⚠️ 파티션의 수는 키를 활용한 메세지 처리에서 중요하기 때문에 만약 토픽에 파티션을 추가해야 한다면, 이것 때문에 토픽을 읽고 있는 어플리케이션이 깨지지 않는 지 확인할 필요가 있다.
토픽에서 레코드 삭제하기
토픽에 30일간의 보존 기한이 설정되어 있다 하더라도 파티션별로 모든 데이터가 하나의 세그먼트에 저장되어 있다면 보존 기한을 넘긴 데이터라 한들 삭제되지 않을 수도 있다. 이때는 특정 시간 이전의 쓰여진 레코드를 지울 필요가 있다.
public static void main(String[] args) throws ExecutionException, InterruptedException {
Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> olderOffsets =
adminClient.listOffsets(reqOrderOffsets).all().get();
Map<TopicPartition, RecordsToDelete> recordsToDelete = new HashMap<>();
for (Map.Entry<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> e : olderOffsets.entrySet()) {
recordsToDelete.put(e.getKey(), RecordsToDelete.beforeOffset(e.getValue().offset()));
}
adminClient.deleteRecords(recordsToDelete).all().get();
}
deleteRecords()는 지정된 오프셋보다 더 오래된 모든 레코드에 삭제 표시를 함으로써, 컨슈머가 접근할 수 없도록 한다.
파티션 0번에 새 레플리카를 새 브로커(ID = 1)에 추가한다. 단 리더는 변경하지 않는다.
파티션 1번에는 레플리카 추가 없이 기존에 있던 레플리카를 가져온다. 레플리카가 하나뿐인 만큼 이것이 리더가 된다.
파티션 2번에 새로운 레플리카를 추가하고, 이것을 선호 리더로 선정한다.
파티션 3번에는 재할당을 하지 않는다.
이렇게 만든 파티션 추가 할당 전략을 alterPartitionReassignments()를 통해 적용한다.
8️⃣ 테스트 하기
AdminClient를 목업(mock-up)한 MockAdminClient클래스는 실제 어드민 작업을 수행할 필요 없이 어플리케이션이 제대도 동작하는지 확인할때 사용한다. API의 일부가 아닌 만큼 언제나 경고 없이 변경될 수 있지만, 공개된 메서드에 대한 목업이기에 메서드 시그니처는 유지된다.
모든 메서드에 대한 목업이 존재하는 것이 아니며, 변경으로 인해 언제든 테스트가 깨질 수 있다는 점을 유의하자.
예시
class Test {
private AdminClient adminClient;
@BeforeEach
void setUp() {
Node broker = new Node(0, "localhost", 9092);
this.adminClient = spy(new MockAdminClient(List.of(broker), broker));
AlterConfigsResult emptyResult = mock(AlterConfigsResult.class);
doReturn(KafkaFuture.completedFuture(null)).when(emptyResult).all();
doReturn(emptyResult).when(adminClient).incrementalAlterConfigs(any());
}
@Test
public void testCreateTopic() throws ExecutionException, InterruptedException {
TopicCreator tc = new TopicCreator(adminClient);
tc.maybeCreateTopic("test.is.a.test.topic");
verify(adminClient, times(1)).createTopics(any());
}
@Test
public void tetNotTopic() throws ExecutionException, InterruptedException {
TopicCreator tc = new TopicCreator(adminClient);
tc.maybeCreateTopic("not.a.test");
verify(adminClient, never()).createTopics(any());
}
}
'test'라는 이름을 받을 경우, 토픽을 생성하는 `TopicCreator` 클래스를 테스트하는 예시이다. 핵심만 살펴보자.
어드민 클라이언트 생성시, `new MockAdminClient()`로 목업 어드민 클라이언트를 할당해주고 있다.
TopicCreator클래서에서 토픽 설정을 변경하는 과정이 있는데, 목업 클라이언트에 해당 메서드가 구현이 안되어 있어서incrementalAlterConfigs을 스터빙(stubbing)을 통해 해결하고 있다. 이렇게 목업이 되어있지 않은 메서드의 경우, Mockito와 같은 테스트 프레임워크를 사용하면 된다.
MockAdminClient를 test jar에 담아서 공개하기 때문에 pom.mxl 의존성을 추가하는 것을 잊지말자.
