Pulsar 커넥터 디버깅하기
Pulsar 커넥터 디버깅하기 (How to debug Pulsar connectors)
이 가이드는 localrun 또는 클러스터 모드에서 커넥터를 디버깅하는 방법과 디버깅 체크리스트를 설명해요. Pulsar 커넥터를 디버깅하는 방법을 더 잘 보여주기 위해 Mongo sink 커넥터를 예시로 사용할게요.
출처: 문서
본문
Mongo sink 환경 배포하기 (Deploy a Mongo sink environment)
- Mongo 서비스를 시작해요.
docker pull mongo:4
docker run -d -p 27017:27017 --name pulsar-mongo -v $PWD/data:/data/db mongo:4
- DB와 컬렉션을 만들어요.
docker exec -it pulsar-mongo /bin/bash
mongo> use pulsar
> db.createCollection('messages')
> exit
- Pulsar standalone을 시작해요.
docker pull apachepulsar/pulsar:2.4.0
docker run -d -it -p 6650:6650 -p 8080:8080 -v $PWD/data:/pulsar/data --link pulsar-mongo --name pulsar-mongo-standalone apachepulsar/pulsar:2.4.0 bin/pulsar standalone
mongo-sink-config.yaml파일로 Mongo sink를 구성해요.
configs:
mongoUri: "mongodb://pulsar-mongo:27017"
database: "pulsar"
collection: "messages"
batchSize: 2
batchTimeMs: 500
docker cp mongo-sink-config.yaml pulsar-mongo-standalone:/pulsar/
- Mongo sink nar 패키지를 다운로드해요.
docker exec -it pulsar-mongo-standalone /bin/bash
curl -O http://apache.01link.hk/pulsar/pulsar-2.4.0/connectors/pulsar-io-mongo-2.4.0.nar
localrun 모드에서 디버깅하기 (Debug in localrun mode)
localrun 명령으로 Mongo sink를 localrun 모드에서 시작해요.
팁:
localrun명령에 대한 더 자세한 내용은localrun문서를 참고하세요.
./bin/pulsar-admin sinks localrun \
--archive $PWD/connectors/pulsar-io-mongo-5.0.0-M2.nar \
--tenant public --namespace default \
--inputs test-mongo \
--name pulsar-mongo-sink \
--sink-config-file $PWD/mongo-sink-config.yaml \
--parallelism 1
커넥터 로그 사용하기 (Use connector log)
localrun 모드에서 커넥터를 디버깅하려면 다음 방법 중 하나로 커넥터 로그를 얻을 수 있어요.
localrun명령 실행 후 로그가 콘솔에 자동으로 출력돼요.- 로그는 다음 위치에 있어요.
logs/functions/tenant/namespace/function-name/function-name-instance-id.log
예시:
Mongo sink 커넥터의 경로는 다음과 같아요.
logs/functions/public/default/pulsar-mongo-sink/pulsar-mongo-sink-0.log
로그 정보를 명확히 설명하기 위해, 아래에서 더 작은 블록으로 나누고 설명을 추가했어요.
- 이 로그 정보는 nar 패키지 압축 해제 후의 저장 경로를 보여줘요.
08:21:54.132 [main] INFO org.apache.pulsar.common.nar.NarClassLoader - Created class loader with paths: [file:/tmp/pulsar-nar/pulsar-io-mongo-2.4.0.nar-unpacked/, file:/tmp/pulsar-nar/pulsar-io-mongo-2.4.0.nar-unpacked/META-INF/bundled-dependencies/,
팁:
class cannot be found예외가 발생하면, nar 파일이file:/tmp/pulsar-nar/pulsar-io-mongo-2.4.0.nar-unpacked/META-INF/bundled-dependencies/폴더에서 압축이 풀렸는지 확인하세요.
- 이 로그 정보는 tenant, namespace, name, parallelism, resources 등 Mongo sink 커넥터의 기본 정보를 보여주며, Mongo sink 커넥터가 올바르게 구성됐는지 확인하는 데 사용할 수 있어요.
08:21:55.390 [main] INFO org.apache.pulsar.functions.runtime.ThreadRuntime - ThreadContainer starting function with instance config InstanceConfig(instanceId=0, functionId=853d60a1-0c48-44d5-9a5c-6917386476b2, functionVersion=c2ce1458-b69e-4175-88c0-a0a856a2be8c, functionDetails=tenant: "public"namespace: "default"name: "pulsar-mongo-sink"className: "org.apache.pulsar.functions.api.utils.IdentityFunction"autoAck: trueparallelism: 1source {typeClassName: "[B"inputSpecs { key: "test-mongo" value { }}cleanupSubscription: true}sink {className: "org.apache.pulsar.io.mongodb.MongoSink"configs: "{\"mongoUri\":\"mongodb://pulsar-mongo:27017\",\"database\":\"pulsar\",\"collection\":\"messages\",\"batchSize\":2,\"batchTimeMs\":500}"typeClassName: "[B"}resources {cpu: 1.0ram: 1073741824disk: 10737418240}componentType: SINK, maxBufferedTuples=1024, functionAuthenticationSpec=null, port=38459, clusterName=local)
- 이 로그 정보는 Mongo에 대한 연결 상태와 구성 정보를 보여줘요.
08:21:56.231 [cluster-ClusterId{value='5d6396a3c9e77c0569ff00eb', description='null'}-pulsar-mongo:27017] INFO org.mongodb.driver.connection - Opened connection [connectionId{localValue:1, serverValue:8}] to pulsar-mongo:27017
08:21:56.326 [cluster-ClusterId{value='5d6396a3c9e77c0569ff00eb', description='null'}-pulsar-mongo:27017] INFO org.mongodb.driver.cluster - Monitor thread successfully connected to server with description ServerDescription{address=pulsar-mongo:27017, type=STANDALONE, state=CONNECTED, ok=true, version=ServerVersion{versionList=[4, 2, 0]}, minWireVersion=0, maxWireVersion=8, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=30, roundTripTimeNanos=89058800}
- 이 로그 정보는 토픽 이름, 구독 이름, 구독 타입 등 컨슈머와 클라이언트의 구성을 설명해요.
08:21:56.719 [pulsar-client-io-1-1] INFO org.apache.pulsar.client.impl.ConsumerStatsRecorderImpl - Starting Pulsar consumer status recorder with config: {"topicNames" : [ "test-mongo" ],"topicsPattern" : null,"subscriptionName" : "public/default/pulsar-mongo-sink","subscriptionType" : "Shared","receiverQueueSize" : 1000,"acknowledgementsGroupTimeMicros" : 100000,"negativeAckRedeliveryDelayMicros" : 60000000,"maxTotalReceiverQueueSizeAcrossPartitions" : 50000,"consumerName" : null,"ackTimeoutMillis" : 0,"tickDurationMillis" : 1000,"priorityLevel" : 0,"cryptoFailureAction" : "CONSUME","properties" : { "application" : "pulsar-sink", "id" : "public/default/pulsar-mongo-sink", "instance_id" : "0"},"readCompacted" : false,"subscriptionInitialPosition" : "Latest","patternAutoDiscoveryPeriod" : 1,"regexSubscriptionMode" : "PersistentOnly","deadLetterPolicy" : null,"autoUpdatePartitions" : true,"replicateSubscriptionState" : false,"resetIncludeHead" : false}
08:21:56.726 [pulsar-client-io-1-1] INFO org.apache.pulsar.client.impl.ConsumerStatsRecorderImpl - Pulsar client config: {"serviceUrl" : "pulsar://localhost:6650","authPluginClassName" : null,"authParams" : null,"operationTimeoutMs" : 30000,"statsIntervalSeconds" : 60,"numIoThreads" : 1,"numListenerThreads" : 1,"connectionsPerBroker" : 1,"useTcpNoDelay" : true,"useTls" : false,"tlsTrustCertsFilePath" : null,"tlsAllowInsecureConnection" : false,"tlsHostnameVerificationEnable" : false,"concurrentLookupRequest" : 5000,"maxLookupRequest" : 50000,"maxNumberOfRejectedRequestPerConnection" : 50,"keepAliveIntervalSeconds" : 30,"connectionTimeoutMs" : 10000,"requestTimeoutMs" : 60000,"defaultBackoffIntervalNanos" : 100000000,"maxBackoffIntervalNanos" : 30000000000}
클러스터 모드에서 디버깅하기 (Debug in cluster mode)
클러스터 모드에서 커넥터를 디버깅하려면 다음 방법을 사용할 수 있어요.
- 커넥터 로그 사용
- admin CLI 사용
커넥터 로그 사용하기 (Use connector log)
클러스터 모드에서는 여러 커넥터가 워커(worker)에서 실행될 수 있어요. 지정된 커넥터의 로그 경로를 찾으려면 workerId를 사용해 커넥터 로그를 찾아요.
admin CLI 사용하기 (Use admin CLI)
Pulsar admin CLI는 다음 하위 명령어로 Pulsar 커넥터를 디버깅하는 데 도움을 줘요.
getstatustopics stats
Mongo sink 만들기
./bin/pulsar-admin sinks create \
--archive $PWD/pulsar-io-mongo-2.4.0.nar \
--tenant public \
--namespace default \
--inputs test-mongo \
--name pulsar-mongo-sink \
--sink-config-file $PWD/mongo-sink-config.yaml \
--parallelism 1
get
get 명령으로 tenant, namespace, name, parallelism 등 Mongo sink 커넥터의 기본 정보를 가져와요.
./bin/pulsar-admin sinks get --tenant public --namespace default --name pulsar-mongo-sink
출력:
{
"tenant": "public",
"namespace": "default",
"name": "pulsar-mongo-sink",
"className": "org.apache.pulsar.io.mongodb.MongoSink",
"inputSpecs": {
"test-mongo": {
"isRegexPattern": false
}
},
"configs": {
"mongoUri": "mongodb://pulsar-mongo:27017",
"database": "pulsar",
"collection": "messages",
"batchSize": 2.0,
"batchTimeMs": 500.0
},
"parallelism": 1,
"processingGuarantees": "ATLEAST_ONCE",
"retainOrdering": false,
"autoAck": true
}
팁:
get명령에 대한 더 자세한 내용은get문서를 참고하세요.
status
status 명령으로 인스턴스 수, 실행 중인 인스턴스 수, instanceId, workerId 등 Mongo sink 커넥터의 현재 상태를 가져와요.
./bin/pulsar-admin sinks status --tenant public \--namespace default \--name pulsar-mongo-sink
출력:
{"numInstances" : 1,"numRunning" : 1,"instances" : [ { "instanceId" : 0, "status" : {
"running" : true, "error" : "", "numRestarts" : 0, "numReadFromPulsar" : 0, "numSystemExceptions" : 0, "latestSystemExceptions" : [ ], "numSinkExceptions" : 0, "latestSinkExceptions" : [ ], "numWrittenToSink" : 0, "lastReceivedTime" : 0, "workerId" : "c-standalone-fw-5d202832fd18-8080" }} ]}
팁:
status명령에 대한 더 자세한 내용은status문서를 참고하세요. 워커에 여러 커넥터가 실행 중이면workerId로 지정된 커넥터가 실행 중인 워커를 찾을 수 있어요.
topics stats
topics stats 명령으로 토픽과 그에 연결된 프로듀서/컨슈머의 통계를 가져와요. 예를 들어 토픽이 메시지를 받았는지, 메시지 백로그가 있는지, 사용 가능한 permits 등 주요 정보를 확인할 수 있어요. 모든 비율은 1분 창(window)에 걸쳐 계산되며, 마지막으로 완료된 1분 기간을 기준으로 해요.
./bin/pulsar-admin topics stats test-mongo
출력:
{
"msgRateIn" : 0.0,
"msgThroughputIn" : 0.0,
"msgRateOut" : 0.0,
"msgThroughputOut" : 0.0,
"averageMsgSize" : 0.0,
"storageSize" : 1,
"publishers" : [ ],
"subscriptions" : {
"public/default/pulsar-mongo-sink" : {
"msgRateOut" : 0.0,
"msgThroughputOut" : 0.0,
"msgRateRedeliver" : 0.0,
"msgBacklog" : 0,
"blockedSubscriptionOnUnackedMsgs" : false,
"msgDelayed" : 0,
"unackedMessages" : 0,
"type" : "Shared",
"msgRateExpired" : 0.0,
"consumers" : [ {
"msgRateOut" : 0.0,
"msgThroughputOut" : 0.0,
"msgRateRedeliver" : 0.0,
"consumerName" : "dffdd",
"availablePermits" : 999,
"unackedMessages" : 0,
"blockedConsumerOnUnackedMsgs" : false,
"metadata" : {
"instance_id" : "0",
"application" : "pulsar-sink",
"id" : "public/default/pulsar-mongo-sink"
},
"connectedSince" : "2019-08-26T08:48:07.582Z",
"clientVersion" : "2.4.0",
"address" : "/172.17.0.3:57790"
} ],
"isReplicated" : false
}
},
"replication" : { },
"deduplicationStatus" : "Disabled"
}
팁:
topic stats명령에 대한 더 자세한 내용은topic stats문서를 참고하세요.
체크리스트 (Checklist)
이 체크리스트는 커넥터를 디버깅할 때 확인해야 할 주요 영역을 나타내요. 철저한 검토를 위해 무엇을 찾아야 하는지 상기시켜 주고, 커넥터의 상태를 파악하는 평가 도구 역할을 해요.
- Pulsar가 성공적으로 시작됐나요?
- 외부 서비스가 정상적으로 동작하나요?
- nar 패키지가 완전한가요?
- 커넥터 구성 파일이 올바른가요?
- localrun 모드에서는, 커넥터를 실행하고 콘솔에 출력되는 정보(커넥터 로그)를 확인하세요.
- 클러스터 모드에서는:
get명령으로 기본 정보를 가져오세요.status명령으로 현재 상태를 가져오세요.topics stats명령으로 지정된 토픽과 그에 연결된 프로듀서/컨슈머의 통계를 가져오세요.- 커넥터 로그를 확인하세요.
- 외부 시스템에 들어가 결과를 검증하세요.