Pulsar 관리 API 호출하기

Java SDK를 사용하는 Pulsar 함수는 Pulsar admin 클라이언트에 접근할 수 있어요. 이 admin 클라이언트가 있으면 함수 안에서 Pulsar 클러스터를 대상으로 관리 API 호출을 할 수 있죠.

함수 컨텍스트(context)에서 노출된 Pulsar admin 클라이언트를 활용하는 예시를 하나씩 살펴볼게요. 아래 예시는 입력 메시지마다 현재 함수 구독(subscription)의 커서(cursor)를 지정한 타임스탬프로 리셋하는 함수예요.

출처: 문서

본문

아래는 함수 컨텍스트에서 노출된 Pulsar admin 클라이언트를 사용하는 예시예요.

import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;

/**
 * In this particular example, for every input message,
 * the function resets the cursor of the current function's subscription to a
 * specified timestamp.
 */
public class CursorManagementFunction implements Function<String, String> {

    @Override
    public String process(String input, Context context) throws Exception {
        PulsarAdmin adminClient = context.getPulsarAdmin();
        if (adminClient != null) {
            String topic = context.getCurrentRecord().getTopicName().isPresent() ?
                    context.getCurrentRecord().getTopicName().get() : null;
            String subName = context.getTenant() + "/" + context.getNamespace() + "/" + context.getFunctionName();
            if (topic != null) {
                // 1578188166 below is a random-pick timestamp
                adminClient.topics().resetCursor(topic, subName, 1578188166);
                return "reset cursor successfully";
            }
        }
        return null;
    }
}

함수가 Pulsar admin 클라이언트에 접근하려면 conf/functions_worker.yml 파일에서 exposeAdminClientEnabled=true로 설정해야 해요. 활성화 여부를 확인하려면 --web-service-url 플래그를 붙여 pulsar-admin functions localrun 명령을 아래처럼 실행하면 돼요.

bin/pulsar-admin functions localrun \
 --jar $PWD/my-functions.jar \
 --classname my.package.CursorManagementFunction \
 --web-service-url http://pulsar-web-service:8080 \
 # Other function configs

더 알아보기 (Learn more)

함수 컨텍스트에서 제공하는 다른 API가 궁금하다면 함수 개발 API 문서를 참고해요.
관리 API 명령어 전체 목록은 Pulsar admin API 문서에서 확인할 수 있어요.
커서 리셋과 같은 토픽 관리 기능은 토픽 관리 문서를 살펴보세요.