토픽 컴팩션 쿡북

토픽 컴팩션 쿡북

토픽에 메시지가 계속 쌓이면 오래되고 가려진 엔트리가 이력을 읽을 때 부담이 돼요. Pulsar의 토픽 컴팩션 기능을 쓰면 컴팩션된 토픽을 만들어 오래된 "가려진" 엔트리를 토픽에서 정리하고, 토픽 이력을 더 빠르게 읽을 수 있어요. 이 글에서는 컴팩션을 실제로 구성하고 실행하는 방법을 단계별로 정리해 드릴게요.

출처: 문서

본문

Pulsar의 토픽 컴팩션 기능을 사용하면 컴팩션된 토픽을 만들 수 있어요. 컴팩션된 토픽에서는 오래되고 "가려진" 엔트리가 토픽에서 정리되어 토픽 이력을 더 빠르게 읽을 수 있어요 (어떤 메시지가 가려졌다/오래됐다/무관하다고 간주되는지는 사용 사례에 따라 달라져요).

컴팩션을 사용하려면:

• 메시지에 키를 부여해야 해요. Pulsar의 토픽 컴팩션은 키(key)별로 일어나기 때문이에요 (즉 메시지는 키에 따라 컴팩션돼요). 주식 티커 사용 사례에서는 주식 심볼(예: AAPL 또는 GOOG)이 키 역할을 할 수 있어요 (아래에서 더 자세히 설명해요). 키가 없는 메시지는 컴팩션 과정에서 그대로 남겨져요.
• 컴팩션을 자동으로 실행하도록 구성하거나, Pulsar 관리 API로 컴팩션을 수동으로 트리거할 수 있어요.
• 소비자가 컴팩션된 토픽에서 읽도록 구성해야 해요 (예를 들어 Java 소비자에는 true로 설정해야 하는 readCompacted 설정이 있어요). 이 구성을 설정하지 않으면 소비자는 여전히 컴팩션되지 않은 토픽에서 읽을 수 있어요.

팁 컴팩션은 키가 있는 메시지에만 동작해요 (주식 티커 예시처럼 각 메시지의 주식 심볼이 키 역할을 하는 방식이에요). 키는 컴팩션이 적용되는 축으로 생각할 수 있어요. 키가 없는 메시지는 컴팩션이 그냥 무시해요.

PIP-318은 브로커 측의 topicCompactionRetainNullKey 구성을 도입했는데, 이를 통해 컴팩션 중 키가 없는 메시지를 보존할지 여부를 구성할 수 있어요. 그리고 3.2.0+ 버전부터는 기본적으로 토픽 컴팩션 중 null 키 메시지를 보존하지 않아요. 자세한 내용은 PIP-318을 보세요.

컴팩션된 토픽은 언제 사용해야 하나요?

컴팩션의 이점을 누릴 수 있는 대표적인 예시는 주식 티커 토픽이에요. 주식 티커 토픽에서 소비자는 특정 주식의 최신 값을 접근할 수 있어요. 메시지가 주식 값을 담고 있고 주식 심볼을 키로 사용한다고(GOOG, AAPL, TWTR 등) 상상해 보세요. 이 토픽을 컴팩션하면 토픽의 소비자에게 두 가지 옵션이 생겨요:

• "역사적" 값, 즉 토픽 메시지 전체에 접근해야 하는 경우 "원본" 비컴팩션 토픽에서 읽을 수 있어요.
• 가장 최신 메시지만 보고 싶다면 컴팩션된 토픽에서 읽을 수 있어요.

따라서 stock-values라는 Pulsar 토픽을 사용한다면, 일부 소비자는 토픽의 모든 메시지에 접근할 수 있고(지난 1시간의 모든 값을 계산하는 작업을 수행하기 때문일 수 있어요), 실시간 주식 티커를 구동하는 소비자만 컴팩션된 토픽을 보게 돼요 (그래서 오래된 메시지를 처리할 필요가 없어요). 특정 소비자가 토픽의 어느 변형에서 메시지를 가져오는지는 소비자의 구성에 따라 결정돼요.

Pulsar 컴팩션의 이점 중 하나는 컴팩션과 비컴팩션 토픽 중에서 고르라고 강요하지 않는다는 거예요. 컴팩션 과정이 원본 토픽을 그대로 두고 본질적으로 대체 토픽을 추가하기 때문이에요. 다시 말해 토픽에 컴팩션을 실행해도, 비컴팩션 버전의 토픽에 접근해야 하는 소비자는 불리한 영향을 받지 않아요.

컴팩션을 자동으로 실행하도록 구성

컴팩션 정책은 컴팩션이 트리거되기 전에 토픽 백로그가 얼마나 커질 수 있는지 지정해요.

테넌트 관리자는 네임스페이스 또는 토픽 수준에서 컴팩션 정책을 구성할 수 있어요. 네임스페이스 수준에서 컴팩션 정책을 구성하면 그 네임스페이스 안의 모든 토픽에 적용돼요.

예를 들어 백로그가 100MB에 도달하면 네임스페이스에서 컴팩션을 트리거하려면:

bin/pulsar-admin namespaces set-compaction-threshold \
--threshold 100M my-tenant/my-namespace

컴팩션을 수동으로 트리거

토픽에서 컴팩션을 실행하려면 pulsar-admin CLI 도구의 topics compact 명령을 사용해야 해요. 예시:

bin/pulsar-admin topics compact \
persistent://my-tenant/my-namespace/my-topic

pulsar-admin 도구는 Pulsar REST API를 통해 컴팩션을 실행해요. 컴팩션을 전용 프로세스에서 실행하려면, 즉 REST API를 통하지 않고 실행하려면 pulsar compact-topic 명령을 사용할 수 있어요. 예시:

bin/pulsar compact-topic \
--topic persistent://my-tenant-namespace/my-topic

컴팩션을 자체 프로세스에서 실행하는 것은 브로커 성능을 방해하지 않으려 할 때 권장돼요. 하지만 브로커 성능은 키스페이스가 큰 토픽(토픽에 키가 많은 경우)에서 컴팩션을 실행할 때만 영향을 받아야 해요. 컴팩션 과정의 첫 번째 단계는 토픽의 각 키 복사본을 유지하는데, 키 수가 늘어나면 메모리 압력이 생길 수 있어요. pulsar-admin topics compact 명령으로 REST API를 통해 컴팩션을 실행하는 것은 대부분의 경우 문제가 없어요. pulsar compact-topic을 사용하는 것은 그에 따라 엣지 케이스로 간주해야 해요.

pulsar compact-topic 명령은 ZooKeeper와 직접 통신해요. ZooKeeper와 통신하려면 pulsar CLI 도구가 유효한 브로커 구성을 갖고 있어야 해요. conf/broker.conf에 적절한 구성을 제공하거나 구성의 기본 위치가 아닌 위치를 지정할 수 있어요:

bin/pulsar compact-topic \
--broker-conf /path/to/broker.conf \
--topic persistent://my-tenant/my-namespace/my-topic
# If the configuration is in conf/broker.conf
bin/pulsar compact-topic \
--topic persistent://my-tenant/my-namespace/my-topic

팁 토픽 컴팩션을 트리거하는 빈도는 사용 사례에 따라 크게 달라져요. 컴팩션된 토픽이 읽기에서 아주 빠르길 원한다면 컴팩션을 꽤 자주 실행해야 해요.

소비자 구성

Pulsar 소비자와 리더는 컴팩션된 토픽에서 읽도록 구성해야 해요. 아래 섹션은 Java 클라이언트에서 컴팩션된 토픽 읽기를 활성화하는 방법을 소개해요.

Java 소비자로 컴팩션된 토픽에서 읽으려면 readCompacted 파라미터를 true로 설정해야 해요. 컴팩션된 토픽용 소비자 예시:

Consumer<byte[]> compactedTopicConsumer = client.newConsumer()
        .topic("some-compacted-topic")
        .readCompacted(true)
        .subscribe();

위에서 언급했듯이 Pulsar의 토픽 컴팩션은 키 기반으로 동작해요. 즉 컴팩션된 토픽에서 생산하는 메시지는 키를 가져야 해요 (키의 내용은 사용 사례에 따라 달라져요). 키가 없는 메시지는 컴팩션 과정에서 무시돼요. 키가 있는 Pulsar 메시지 예시:

import org.apache.pulsar.client.api.TypedMessageBuilder;
TypedMessageBuilder<byte[]> msg = producer.newMessage()
        .key("some-key")
        .value(someByteArray);

아래 예시는 컴팩션된 Pulsar 토픽에 키가 있는 메시지를 생산하는 모습을 보여줘요:

import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
PulsarClient client = PulsarClient.builder()
        .serviceUrl("pulsar://localhost:6650")
        .build();
Producer<byte[]> compactedTopicProducer = client.newProducer()
        .topic("some-compacted-topic")
        .create();
compactedTopicProducer.newMessage()
        .key("some-key")
        .value(someByteArray)
        .send();

토픽에서 컴팩션 제거하기

버전 2.11.4, 3.0.3, 3.1.0부터 (PR 21745 덕분에) __compaction 구독을 제거함으로써 컴팩션을 제거할 수 있어요.

bin/pulsar-admin topics unsubscribe {topic} -s __compaction

버전 2.11.4, 3.0.3, 3.1.0 이전에는 토픽에서 컴팩션을 제거하려면 컴팩션 데이터를 제거할 BookKeeper 셸 접근이 필요했어요. 단계는 다음과 같아요:

1. stats-internal의 compactedLedger에서 컴팩션된 ledgerId를 가져와요.

bin/pulsar-admin topics stats-internal {topic}

2. 컴팩션된 ledger를 삭제해요.

bin/bookkeeper shell deleteledger {compacted_ledger}

3. 토픽을 언로드해 브로커의 캐시를 비워요.

bin/pulsar-admin topics unload {topic}

더 알아보기 (Learn more)

• 토픽 컴팩션 개념 — 컴팩션이 왜 필요한지와 동작 원리를 이해해요.
• pulsar-admin CLI 참조 — topics, namespaces 명령을 살펴보세요.
• 메시징 개념 — 키와 페이로드의 기본을 익혀요.