Pulsar 관리 API 시작하기

Pulsar 관리 API 시작하기 (Get started)

Pulsar 관리 인터페이스는 클러스터를 관리할 수 있게 해줘요. Pulsar 인스턴스의 모든 엔티티(클러스터, 테넌트, 네임스페이스, 토픽, 스키마, 함수, 커넥터, 패키지, 트랜잭션)를 생성·업데이트·삭제·점검하고 데이터·리소스·보안에 대한 정책을 설정할 수 있어요. Pulsar를 관리하는 모든 방법은 결국 브로커가 제공하는 관리 REST API를 거치는데, 커맨드라인 도구와 클라이언트 라이브러리는 모두 그것의 클라이언트예요.

출처: 문서

본문

Pulsar를 관리하는 모든 방법은 브로커가 서빙하는 관리 REST API를 거쳐요. 커맨드라인 도구와 클라이언트 라이브러리는 모두 그 REST API의 클라이언트예요. 작업 방식에 맞는 인터페이스를 고르세요.

Interface	Reference	Use it when
Command line: pulsar-admin	pulsar-admin CLI reference	대화형으로 또는 셸 스크립트에서 클러스터를 관리해요. Pulsar 설치본의 bin 폴더에 포함되어 있어 추가 설정이 필요 없지만, 호출마다 JVM이 시작돼요. Pulsar shell은 같은 명령을 감싼 대화형 셸로, JVM을 한 번만 시작해요.
REST API	REST APIs and OpenAPI specifications, endpoint reference	HTTP를 말하는 어떤 언어·도구로든 자동화해요. 예를 들어 운영자, 컨트롤러, 스크립트의 curl 호출이죠. Pulsar 5.0부터 API는 OpenAPI 3 스펙을 갖고 있어 그로부터 언어용 클라이언트를 생성할 수 있어요.
Java Library	Java admin API Javadoc	Java 애플리케이션이나 테스트에서 Pulsar를 관리해요. 클라이언트는 org.apache.pulsar:pulsar-client-admin 아티팩트예요.
Go Library	pulsaradmin package of the Go client	Go 애플리케이션에서 Pulsar를 관리해요. 예를 들어 Kubernetes 오퍼레이터가 있죠.

REST API는 더 자세히 볼 가치가 있어요. 다른 인터페이스가 그 위에 구축되어 있고, CLI나 Java 클라이언트가 맞지 않을 때 적절한 선택이기 때문이에요.

• 직접 호출하기: 기능 페이지의 모든 연산에 해당 엔드포인트가 있는 REST API 탭이 있어요. 패키지(functions, sources, sinks)를 업로드하는 엔드포인트는 multipart 요청을 받아요. 함수 생성 예제의 curl 예시를 참고하세요.
• OpenAPI 스펙 사용하기: Pulsar 5.0부터 REST API는 릴리스마다 브로커 소스에서 생성된 OpenAPI 3 문서로 기술돼요. API별로(admin, lookup, functions, sources, sinks, packages, transactions) 하나씩이에요. 릴리스 라인별로 다운로드해서 Postman 같은 도구에 가져올 수 있고, REST API 참고서를 렌더링하는 데도 쓰여요.
• 언어용 클라이언트 생성하기: 스펙과 OpenAPI Generator로 Go, Python, TypeScript, Rust, C# 등 다양한 언어의 클라이언트 라이브러리를 직접 작성하는 대신 생성할 수 있어요.

이 페이지의 나머지는 커맨드라인, REST API, Java API로 각각 토픽을 관리하면서 가장 빠르게 시작하는 방법을 안내해요. 각 인터페이스가 인증이 활성화되었을 때 필요한 설정도 포함해요.

pulsar-admin

pulsar-admin CLI는 커맨드라인 도구이며 Pulsar 설치본의 bin 폴더에서 사용할 수 있어요.

REST API

REST API는 브로커가 제공하는 관리 API에 대한 HTTP 호출이에요. Java admin API와 pulsar-admin CLI도 모두 REST API를 사용해요. 이 API는 OpenAPI 스펙으로 기술되며, 그 스펙으로 다른 언어용 클라이언트를 생성할 수 있어요.

Java

Java admin API는 Java로 작성된 프로그램 가능한 인터페이스예요.

자세한 단계는 아래를 확인해요.

pulsar-admin CLI로 토픽 관리하기

pulsar-admin CLI로 토픽을 관리하려면 다음 단계를 완료해요.

1. 서비스 URL을 설정한다.
2. 파티션 토픽을 생성한다.
3. 파티션 수를 업데이트한다.
4. 토픽에 메시지를 생성(produce)한다.
5. 토픽의 통계를 확인한다.
6. 토픽을 삭제한다.

전제 조건(Prerequisites)

Pulsar standalone을 설치하고 시작해요. 이 튜토리얼은 예시로 Pulsar 2.11을 실행해요.

단계(Steps)

Step 1: client.conf에서 서비스 URL이 브로커 서비스를 가리키도록 설정해요.

webServiceUrl=http://localhost:8080/
brokerServiceUrl=pulsar://localhost:6650/

클러스터에서 인증이 활성화되어 있다면 같은 파일에 인증 플러그인과 파라미터를, 웹 서비스가 TLS를 사용한다면 TLS 설정을 구성해요.

Name	Description	Default
webServiceUrl	클러스터의 웹 URL	http://localhost:8080/
brokerServiceUrl	클러스터의 Pulsar 프로토콜 URL	pulsar://localhost:6650/
authPlugin	인증 플러그인
authParams	클러스터의 인증 파라미터 (쉼표로 구분된 문자열)
useTls	클러스터에서 TLS 인증을 강제할지 여부	false
tlsAllowInsecureConnection	클라이언트의 신뢰되지 않은 TLS 인증서 허용	false
tlsTrustCertsFilePath	신뢰된 TLS 인증서 파일 경로

모든 설정은 클라이언트 구성 참고서를 참고해요.

Step 2: test-topic-1이라는 이름의 영속 토픽을 6개 파티션으로 생성해요.

입력:

bin/pulsar-admin topics create-partitioned-topic \
persistent://public/default/test-topic-1 \
--partitions 6

출력: 출력 없음. 토픽 상태는 Step 5에서 확인할 수 있어요.

Step 3: 파티션 수를 8로 업데이트해요.

입력:

bin/pulsar-admin topics update-partitioned-topic \
persistent://public/default/test-topic-1 \
--partitions 8

출력: 출력 없음. 파티션 수는 Step 5에서 확인할 수 있어요.

Step 4: 파티션 토픽 test-topic-1에 몇 개의 메시지를 생성해요.

입력:

bin/pulsar-perf produce -u pulsar://localhost:6650 -r 1000 -i 1000 persistent://public/default/test-topic-1

출력:

2023-03-07T15:33:56,832+0800 [main] INFO  org.apache.pulsar.testclient.PerformanceProducer - Starting Pulsar perf producer with config: {
  "confFile" : "/Users/yu/apache-pulsar-2.11.0/conf/client.conf",
  "serviceURL" : "pulsar://localhost:6650",
  "authPluginClassName" : "",
  "authParams" : "",
  "tlsTrustCertsFilePath" : "",
  "tlsAllowInsecureConnection" : false,
  "tlsHostnameVerificationEnable" : false,
  "maxConnections" : 1,
  "statsIntervalSeconds" : 1000,
  "ioThreads" : 1,
  "enableBusyWait" : false,
  "listenerName" : null,
  "listenerThreads" : 1,
  "maxLookupRequest" : 50000,
  "topics" : [ "persistent://public/default/test-topic-1" ],
  "numTestThreads" : 1,
  "msgRate" : 1000,
  "msgSize" : 1024,
  "numTopics" : 1,"numProducers" : 1,
  "separator" : "-",
  "sendTimeout" : 0,
  "producerName" : null,
  "adminURL" : "http://localhost:8080/",
...
2023-03-07T15:35:03,769+0800 [Thread-0] INFO  org.apache.pulsar.testclient.PerformanceProducer - Aggregated latency stats --- Latency: mean:   8.931 ms - med:   3.775 - 95pct:  32.144 - 99pct:  98.432 - 99.9pct: 216.088 - 99.99pct: 304.807 - 99.999pct: 349.391 - Max: 351.235

Step 5: 파티션 토픽 test-topic-1의 내부 통계를 확인해요.

입력:

bin/pulsar-admin topics partitioned-stats-internal \
persistent://public/default/test-topic-1

출력: 아래는 출력의 일부예요. 토픽 통계에 대한 자세한 설명은 Pulsar statistics를 참고해요.

{
  "metadata" : {
    "partitions" : 8
  },
  "partitions" : {
    "persistent://public/default/test-topic-1-partition-1" : {
      "entriesAddedCounter" : 4213,
      "numberOfEntries" : 4213,
      "totalSize" : 8817693,
      "currentLedgerEntries" : 4212,
      "currentLedgerSize" : 8806289,
      "lastLedgerCreatedTimestamp" : "2023-03-07T15:33:59.367+08:00",
      "waitingCursorsCount" : 0,
      "pendingAddEntriesCount" : 0,
      "lastConfirmedEntry" : "65:4211",
      "state" : "LedgerOpened",
      "ledgers" : [ {
        "ledgerId" : 49,
        "entries" : 1,
        "size" : 11404,
        "offloaded" : false,
        "underReplicated" : false
      }, {
        "ledgerId" : 65,
        "entries" : 0,
        "size" : 0,
        "offloaded" : false,
        "underReplicated" : false
      } ],
      "cursors" : {
        "test-subscriptio-1" : {
          "markDeletePosition" : "49:-1",
          "readPosition" : "49:0",
          "waitingReadOp" : false,
          "pendingReadOps" : 0,
          "messagesConsumedCounter" : 0,
          "cursorLedger" : -1,
          "cursorLedgerLastEntry" : -1,  "individuallyDeletedMessages" : "[]",
          "lastLedgerSwitchTimestamp" : "2023-03-06T16:41:32.801+08:00",
          "state" : "NoLedger",
          "numberOfEntriesSinceFirstNotAckedMessage" : 1,
          "totalNonContiguousDeletedMessagesRange" : 0,
          "subscriptionHavePendingRead" : false,
          "subscriptionHavePendingReplayRead" : false,
          "properties" : { }
        },
        "test-subscription-1" : {
          "markDeletePosition" : "49:-1",
          "readPosition" : "49:0",
          "waitingReadOp" : false,
          "pendingReadOps" : 0,
          "messagesConsumedCounter" : 0,
          "cursorLedger" : -1,
          "cursorLedgerLastEntry" : -1,
          "individuallyDeletedMessages" : "[]",
          "lastLedgerSwitchTimestamp" : "2023-03-06T16:41:32.801+08:00",
          "state" : "NoLedger",
          "numberOfEntriesSinceFirstNotAckedMessage" : 1,
          "totalNonContiguo... [truncated]

Step 6: 토픽 test-topic-1을 삭제해요.

입력:

bin/pulsar-admin topics delete-partitioned-topic persistent://public/default/test-topic-1

출력: 출력 없음. 다음 명령으로 test-topic-1이 존재하는지 여부를 확인할 수 있어요.

입력 — public/default 네임스페이스의 토픽을 나열해요.

bin/pulsar-admin topics list public/default

REST API로 토픽 관리하기

REST API로 토픽을 관리하려면 다음 단계를 완료해요.

1. 파티션 토픽 생성
2. 파티션 수 업데이트
3. 토픽에 메시지 생성
4. 토픽 통계 확인
5. 토픽 삭제

전제 조건(Prerequisites)

Pulsar standalone을 설치하고 시작해요. 이 튜토리얼은 예시로 Pulsar 2.11을 실행해요.

단계(Steps)

Step 1: test-topic-2라는 이름의 영속 토픽을 4개 파티션으로 생성해요.

입력:

curl -X PUT http://localhost:8080/admin/v2/persistent/public/default/test-topic-2/partitions -H 'Content-Type: application/json' -d "4"

인증이 활성화되어 있다면 모든 요청에 인증 제공자의 자격 증명을 추가해요. 예를 들어 -H "Authorization: Bearer *** token)"로 토큰을 넣는 것이죠. TLS를 쓴다면 https:// 웹 서비스 URL(기본 8443)을 사용해요. 엔드포인트는 REST API 참고서에 문서화되어 있고, REST APIs and OpenAPI specifications 페이지에서 API 배치와 클라이언트 생성 방법을 설명해요.

출력: 출력 없음. 토픽은 Step 4에서 확인할 수 있어요.

Step 2: 파티션 수를 5로 업데이트해요.

입력:

curl -X POST http://localhost:8080/admin/v2/persistent/public/default/test-topic-2/partitions -H 'Content-Type: application/json' -d "5"

출력: 출력 없음. 토픽 상태는 Step 4에서 확인할 수 있어요.

Step 3: 파티션 토픽 test-topic-2에 몇 개의 메시지를 생성해요.

입력:

bin/pulsar-perf produce -u pulsar://localhost:6650 -r 1000 -i 1000 persistent://public/default/test-topic-2

출력:

2023-03-08T15:47:06,268+0800 [main] INFO  org.apache.pulsar.testclient.PerformanceProducer - Starting Pulsar perf producer with config: {
  "confFile" : "/Users/yu/apache-pulsar-2.11.0/conf/client.conf",
  "serviceURL" : "pulsar://localhost:6650",
  "authPluginClassName" : "",
  "authParams" : "",
  "tlsTrustCertsFilePath" : "",
  "tlsAllowInsecureConnection" : false,
  "tlsHostnameVerificationEnable" : false,
  "maxConnections" : 1,
  "statsIntervalSeconds" : 1000,
  "ioThreads" : 1,
  "enableBusyWait" : false,
  "listenerName" : null,
  "listenerThreads" : 1,
  "maxLookupRequest" : 50000,
  "topics" : [ "persistent://public/default/test-topic-2" ],
  "numTestThreads" : 1,
  "msgRate" : 1000,
  "msgSize" : 1024,
  "numTopics" : 1,"numProducers" : 1,
  "separator" : "-",
  "sendTimeout" : 0,
  "producerName" : null,
  "adminURL" : "http://localhost:8080/",
  "deprecatedAuthPluginClassName" : null,
  "maxOutstanding" : 0,
  "maxPendingMessagesAcrossPartitions" : 0,
  "partitions" : null,
  "numMessages" : 0,
  "compression" : "NONE",
  "payloadFilename" : null,
  "payloadDelimiter" : "\\n",
  "batchTimeMillis" : 1.0,
  "batchMaxMessages" : 1000,
  "batchMaxBytes" : 4194304,
  "testTime" : 0,
  "warmupTimeSeconds" : 1.0,
  "encKeyName" : null,
  "encKeyFile" : null,
  "delay" : 0,
  "exitOnFailure" : false,
  "messageKeyGenerationMode" : null,
  "producerAccessMode" : "Shared",
  "formatPayload" : false,
  "formatterClass" : "org.apache.pulsar.testclient.DefaultMessageFormatter",
  "transactionTimeout" : 10,
  "numMessagesPerTransaction" : 50,
  "isEnableTransaction" : false,
  "isAbortTransaction" : false,
  "histogramFile" : null}
...
2023-03-08T15:53:28,178+0800 [Thread-0] INFO  org.apache.pulsar.testclient.PerformanceProducer - Aggregated latency stats --- Latency: mean:   4.481 ms - med:   2.918 - 95pct:  10.710 - 99pct:  38.928 - 99.9pct: 112.689 - 99.99pct: 154.241 - 99.999pct: 193.249 - Max: 241.717

Step 4: 토픽 test-topic-2의 내부 통계를 확인해요.

입력:

curl -X GET http://localhost:8080/admin/v2/persistent/public/default/test-topic-2/partitioned-internalStats

출력: 토픽 통계에 대한 자세한 설명은 Pulsar statistics를 참고해요.

{"metadata":{"partitions":5},"partitions":{"persistent://public/default/test-topic-2-partition-3":{"entriesAddedCounter":47087,"numberOfEntries":47087,"totalSize":80406959,"currentLedgerEntries":47087,"currentLedgerSize":80406959,"lastLedgerCreatedTimestamp":"2023-03-08T15:47:07.273+08:00","waitingCursorsCount":0,"pendingAddEntriesCount":0,"lastConfirmedEntry":"117:47086","state":"LedgerOpened","ledgers":[{"ledgerId":117,"entries":0,"size":0,"offloaded":false,"underReplicated":false}],"cursors":{},"schemaLedgers":[],"compactedLedger":{"ledgerId":-1,"entries":-1,"size":-1,"offloaded":false,"underReplicated":false}},"persistent://public/default/test-topic-2-partition-2":{"entriesAddedCounter":46995,"numberOfEntries":46995,"totalSize":80445417,"currentLedgerEntries":46995,"currentLedgerSize":80445417,"lastLedgerCreatedTimestamp":"2023-03-08T15:47:07.43+08:00","waitingCursorsCount":0,"pendingAddEntriesCount":0,"lastConfirmedEntry":"118:46994","state":"LedgerOpened","ledgers":[{"ledgerId":118,"entries":0,"size":0,"offloaded":false,"underReplicated":false}],...

Step 5: 토픽 test-topic-2를 삭제해요.

입력:

curl -X DELETE http://localhost:8080/admin/v2/persistent/public/default/test-topic-2/partitions

출력: 출력 없음. 다음 명령으로 test-topic-2가 존재하는지 여부를 확인할 수 있어요.

입력 — public/default 네임스페이스의 토픽을 나열해요.

curl -X GET http://localhost:8080/admin/v2/persistent/public/default

Java admin API로 토픽 관리하기

Java admin API로 토픽을 관리하려면 다음 단계를 완료해요.

1. Pulsar Java 클라이언트를 초기화한다.
2. 파티션 토픽 생성
3. 파티션 수 업데이트
4. 토픽에 메시지 생성
5. 토픽 통계 확인
6. 토픽 삭제

전제 조건(Prerequisites)

Java 프로젝트를 준비하고 POM 파일에 다음 의존성을 추가해요.

<dependency>
      <groupId>org.apache.pulsar</groupId>
      <artifactId>pulsar-client-admin</artifactId>
      <version>5.0.0-M2</version>
  </dependency>

단계(Steps)

Step 1: Java 프로젝트에서 Pulsar Java 클라이언트를 초기화해요.

입력:

String url = "http://localhost:8080";
PulsarAdmin admin = PulsarAdmin.builder()
    .serviceHttpUrl(url)
    .build();

PulsarAdmin 객체는 PulsarAdminBuilder를 통해 생성돼요. 인증이 활성화되어 있다면 인증 플러그인과 파라미터를, 웹 서비스가 TLS를 쓴다면 TLS 설정을 전달해요. 서비스 URL은 쉼표로 구분해 여러 브로커를 나열할 수 있어요.

String url = "https://broker-1.example.com:8443,broker-2.example.com:8443,broker-3.example.com:8443";
// The fully qualified class name of the authentication plugin, and its parameters
String authPluginClassName = "org.apache.pulsar.client.impl.auth.AuthenticationToken";
String authParams = "file:///path/to/token";
boolean tlsAllowInsecureConnection = false;
String tlsTrustCertsFilePath = "/path/to/ca.cert.pem";
PulsarAdmin admin = PulsarAdmin.builder()
    .serviceHttpUrl(url)
    .authentication(authPluginClassName, authParams)
    .tlsTrustCertsFilePath(tlsTrustCertsFilePath)
    .allowTlsInsecureConnection(tlsAllowInsecureConnection)
    .build();

Step 2: test-topic-1이라는 파티션 토픽을 4개 파티션으로 생성해요.

입력:

admin.topics().createPartitionedTopic("persistent://public/default/test-topic-1", 4);

Step 3: 파티션 수를 5로 업데이트해요.

입력:

admin.topics().updatePartitionedTopic("test-topic-1", 5);

Step 4: 토픽 test-topic-1에 몇 개의 메시지를 생성해요.

입력:

PulsarClient client = PulsarClient.builder()
    .serviceUrl("pulsar://localhost:6650")
    .build();
Producer<String> producer = client.newProducer(Schema.STRING)
    .topic("test-topic-1")
    .blockIfQueueFull(true)
    .create();
for (int i = 0; i < 100; ++i) {
    producer.newMessage().value("test").send();
}
producer.close();
client.close();

Step 5: 토픽 test-topic-1의 통계를 확인해요.

입력:

PartitionedTopicStats stats = admin.topics().getPartitionedStats("persistent://public/default/test-topic-1",false);
System.out.println(stats.getMsgInCounter());

출력:

100

Step 6: 토픽 test-topic-1을 삭제해요.

입력:

admin.topics().deletePartitionedTopic("test-topic-1");

관련 토픽 (Related topics)

• 일반적인 관리 작업을 배우려면 Pulsar admin API - Features를 보세요.
• 상세 사용법은 아래 참고서를 확인해요.
  • pulsar-admin CLI
  • Pulsar admin APIs
  • REST API, OpenAPI specifications and generated clients
  • Java admin API

더 알아보기 (Learn more)

• 관리 인터페이스로 할 수 있는 작업 전체 목록은 Pulsar admin API - Features 문서를 참고해요.
• pulsar-admin CLI의 전체 명령은 CLI 참고서에서 확인해요.
• REST API 엔드포인트와 OpenAPI 스펙은 REST API 참고서와 OpenAPI 문서에 있어요.
• Java admin API의 클래스·메서드는 Javadoc을 참고해요.