2022-07-21T15:48:50,001+0000 [send-response-OrderedScheduler-1-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7368), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670140,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254) responseAndRequest content: {throttle_time_ms=0,error_code=0,session_id=0,responses=[{topic=test-02,partition_responses=[{partition_header={partition=0,error_code=0,high_watermark=15670208,last_stable_offset=15670208,log_start_offset=15670208,aborted_transactions=[]},record_set=[(record=DefaultRecord(offset=15670140, timestamp=1658418529896, key=0 bytes, value=2556 bytes)), (record=DefaultRecord(offset=15670141, timestamp=1658418529896, key=0 bytes, value=2554 bytes)), (record=DefaultRecord(offset=15670142, timestamp=1658418529896, key=0 bytes, value=2598 bytes)), (record=DefaultRecord(offset=15670143, timestamp=1658418529896, key=0 bytes, value=2638 bytes)), (record=DefaultRecord(offset=15670144, timestamp=1658418529896, key=0 bytes, value=2667 bytes)), (record=DefaultRecord(offset=15670145, timestamp=1658418529896, key=0 bytes, value=2867 bytes)), (record=DefaultRecord(offset=15670146, timestamp=1658418529896, key=0 bytes, value=2435 bytes)), (record=DefaultRecord(offset=15670147, timestamp=1658418529896, key=0 bytes, value=2697 bytes)), (record=DefaultRecord(offset=15670148, timestamp=1658418529896, key=0 bytes, value=2697 bytes)), (record=DefaultRecord(offset=15670149, timestamp=1658418529896, key=0 bytes, value=2697 bytes)), (record=DefaultRecord(offset=15670150, timestamp=1658418529918, key=0 bytes, value=2604 bytes)), (record=DefaultRecord(offset=15670151, timestamp=1658418529918, key=0 bytes, value=2671 bytes)), (record=DefaultRecord(offset=15670152, timestamp=1658418529904, key=0 bytes, value=3255 bytes)), (record=DefaultRecord(offset=15670153, timestamp=1658418529904, key=0 bytes, value=2120 bytes)), (record=DefaultRecord(offset=15670154, timestamp=1658418529904, key=0 bytes, value=3252 bytes)), (record=DefaultRecord(offset=15670155, timestamp=1658418529904, key=0 bytes, value=2120 bytes)), (record=DefaultRecord(offset=15670156, timestamp=1658418529814, key=0 bytes, value=1905 bytes)), (record=DefaultRecord(offset=15670157, timestamp=1658418529814, key=0 bytes, value=1861 bytes)), (record=DefaultRecord(offset=15670158, timestamp=1658418529814, key=0 bytes, value=1902 bytes)), (record=DefaultRecord(offset=15670159, timestamp=1658418529814, key=0 bytes, value=1902 bytes)), (record=DefaultRecord(offset=15670160, timestamp=1658418529814, key=0 bytes, value=1908 bytes)), (record=DefaultRecord(offset=15670161, timestamp=1658418529814, key=0 bytes, value=1908 bytes)), (record=DefaultRecord(offset=15670162, timestamp=1658418529814, key=0 bytes, value=1913 bytes)), (record=DefaultRecord(offset=15670163, timestamp=1658418529814, key=0 bytes, value=1896 bytes)), (record=DefaultRecord(offset=15670164, timestamp=1658418529814, key=0 bytes, value=1896 bytes)), (record=DefaultRecord(offset=15670165, timestamp=1658418529814, key=0 bytes, value=1896 bytes)), (record=DefaultRecord(offset=15670166, timestamp=1658418529814, key=0 bytes, value=1895 bytes)), (record=DefaultRecord(offset=15670167, timestamp=1658418529814, key=0 bytes, value=1904 bytes)), (record=DefaultRecord(offset=15670168, timestamp=1658418529814, key=0 bytes, value=1913 bytes)), (record=DefaultRecord(offset=15670169, timestamp=1658418529814, key=0 bytes, value=1911 bytes)), (record=DefaultRecord(offset=15670170, timestamp=1658418529814, key=0 bytes, value=1917 bytes)), (record=DefaultRecord(offset=15670171, timestamp=1658418529814, key=0 bytes, value=1915 bytes)), (record=DefaultRecord(offset=15670172, timestamp=1658418529814, key=0 bytes, value=1917 bytes)), (record=DefaultRecord(offset=15670173, timestamp=1658418529814, key=0 bytes, value=1928 bytes)), (record=DefaultRecord(offset=15670174, timestamp=1658418529814, key=0 bytes, value=1913 bytes)), (record=DefaultRecord(offset=15670175, timestamp=1658418529814, key=0 bytes, value=1927 bytes)), (record=DefaultRecord(offset=15670176, timestamp=1658418529814, key=0 bytes, value=1928 bytes)), (record=DefaultRecord(offset=15670177, timestamp=1658418529814, key=0 bytes, value=1929 bytes)), (record=DefaultRecord(offset=15670178, timestamp=1658418529814, key=0 bytes, value=1913 bytes)), (record=DefaultRecord(offset=15670179, timestamp=1658418529814, key=0 bytes, value=1917 bytes)), (record=DefaultRecord(offset=15670180, timestamp=1658418529814, key=0 bytes, value=1923 bytes)), (record=DefaultRecord(offset=15670181, timestamp=1658418529814, key=0 bytes, value=1917 bytes)), (record=DefaultRecord(offset=15670182, timestamp=1658418529814, key=0 bytes, value=1918 bytes)), (record=DefaultRecord(offset=15670183, timestamp=1658418529814, key=0 bytes, value=1881 bytes)), (record=DefaultRecord(offset=15670184, timestamp=1658418529814, key=0 bytes, value=1885 bytes)), (record=DefaultRecord(offset=15670185, timestamp=1658418529814, key=0 bytes, value=1909 bytes)), (record=DefaultRecord(offset=15670186, timestamp=1658418529814, key=0 bytes, value=1915 bytes)), (record=DefaultRecord(offset=15670187, timestamp=1658418529814, key=0 bytes, value=1917 bytes)), (record=DefaultRecord(offset=15670188, timestamp=1658418529814, key=0 bytes, value=1926 bytes)), (record=DefaultRecord(offset=15670189, timestamp=1658418529814, key=0 bytes, value=1909 bytes)), (record=DefaultRecord(offset=15670190, timestamp=1658418529814, key=0 bytes, value=1915 bytes)), (record=DefaultRecord(offset=15670191, timestamp=1658418529814, key=0 bytes, value=1921 bytes)), (record=DefaultRecord(offset=15670192, timestamp=1658418529814, key=0 bytes, value=1909 bytes)), (record=DefaultRecord(offset=15670193, timestamp=1658418529814, key=0 bytes, value=1921 bytes)), (record=DefaultRecord(offset=15670194, timestamp=1658418529814, key=0 bytes, value=1925 bytes)), (record=DefaultRecord(offset=15670195, timestamp=1658418529814, key=0 bytes, value=1880 bytes)), (record=DefaultRecord(offset=15670196, timestamp=1658418529916, key=0 bytes, value=2810 bytes)), (record=DefaultRecord(offset=15670197, timestamp=1658418529916, key=0 bytes, value=2804 bytes))]}]}]}
2022-07-21T15:48:50,001+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Channel writability has changed to: false
2022-07-21T15:48:50,001+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Channel writability has changed to: false
2022-07-21T15:48:50,002+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Channel writability has changed to: true
2022-07-21T15:48:50,002+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Channel writability has changed to: true
2022-07-21T15:48:50,002+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:55254] Received kafka cmd RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7369), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7369), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670198,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254)
2022-07-21T15:48:50,002+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Request RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7369) Fetch request. Size: 1. Each item:
2022-07-21T15:48:50,002+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler -   Fetch request topic:test-02-0 data:(offset=15670198, logStartOffset=-1, maxBytes=1048576).
2022-07-21T15:48:50,002+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,002+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,002+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Fetch for test-02-0: remove tcm to get cursor for fetch offset: 15670198 .
2022-07-21T15:48:50,002+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Get cursor for offset: 15670198 in cache. cache size: 0
2022-07-21T15:48:50,002+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Mark delete success for position: 469:8733
2022-07-21T15:48:50,002+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Topic test-02-0 success read entry: ledgerId: 469, entryId: 8733, size: 1829, ConsumerManager original offset: 15670198, lastEntryPosition: 469:8733, nextOffset: 15670208
2022-07-21T15:48:50,002+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Add cursor back kop-consumer-cursor-persistent://public/default/test-02-partition-0-414-2987-a2bc5bcbd3 for offset: 15670208
2022-07-21T15:48:50,002+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,002+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Request RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7369): read 2 entries but only 2 entries are committed
2022-07-21T15:48:50,002+0000 [send-response-OrderedScheduler-1-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7369), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670198,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254) responseAndRequest content: {throttle_time_ms=0,error_code=0,session_id=0,responses=[{topic=test-02,partition_responses=[{partition_header={partition=0,error_code=0,high_watermark=15670208,last_stable_offset=15670208,log_start_offset=15670208,aborted_transactions=[]},record_set=[(record=DefaultRecord(offset=15670198, timestamp=1658418529831, key=0 bytes, value=2745 bytes)), (record=DefaultRecord(offset=15670199, timestamp=1658418529831, key=0 bytes, value=2577 bytes)), (record=DefaultRecord(offset=15670200, timestamp=1658418529831, key=0 bytes, value=2577 bytes)), (record=DefaultRecord(offset=15670201, timestamp=1658418529831, key=0 bytes, value=2816 bytes)), (record=DefaultRecord(offset=15670202, timestamp=1658418529831, key=0 bytes, value=2799 bytes)), (record=DefaultRecord(offset=15670203, timestamp=1658418529831, key=0 bytes, value=2816 bytes)), (record=DefaultRecord(offset=15670204, timestamp=1658418529831, key=0 bytes, value=2799 bytes)), (record=DefaultRecord(offset=15670205, timestamp=1658418529831, key=0 bytes, value=2816 bytes)), (record=DefaultRecord(offset=15670206, timestamp=1658418529831, key=0 bytes, value=2799 bytes)), (record=DefaultRecord(offset=15670207, timestamp=1658418529919, key=0 bytes, value=1772 bytes))]}]}]}
2022-07-21T15:48:50,003+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:55254] Received kafka cmd RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7370), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7370), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670208,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254)
2022-07-21T15:48:50,003+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Request RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7370) Fetch request. Size: 1. Each item:
2022-07-21T15:48:50,003+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler -   Fetch request topic:test-02-0 data:(offset=15670208, logStartOffset=-1, maxBytes=1048576).
2022-07-21T15:48:50,003+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,003+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,003+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Fetch for test-02-0: remove tcm to get cursor for fetch offset: 15670208 .
2022-07-21T15:48:50,003+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Get cursor for offset: 15670208 in cache. cache size: 0
2022-07-21T15:48:50,003+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Add cursor back kop-consumer-cursor-persistent://public/default/test-02-partition-0-414-2987-a2bc5bcbd3 for offset: 15670208
2022-07-21T15:48:50,003+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,003+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Request RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7370): read 0 entries but only 0 entries are committed
2022-07-21T15:48:50,012+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:44070] Received kafka cmd RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=166304), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=166304), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=3703]}, remoteAddress=/<IP_ADDRESS>:44070)
2022-07-21T15:48:50,012+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,012+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,015+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessagePublishContext - Success write topic: persistent://public/default/test-02-partition-0, producerName KOP-PID-PREFIX--1--1 ledgerId: 469, entryId: 8734 And triggered send callback.
2022-07-21T15:48:50,015+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.storage.ReplicaManager - Complete handle appendRecords.
2022-07-21T15:48:50,016+0000 [send-response-OrderedScheduler-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=166304), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=3703]}, remoteAddress=/<IP_ADDRESS>:44070) responseAndRequest content: {responses=[{topic=test-02,partition_responses=[{partition=0,error_code=0,base_offset=15670208,log_append_time=-1,log_start_offset=-1}]}],throttle_time_ms=0}
2022-07-21T15:48:50,058+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:53926] Received kafka cmd RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=126180), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=126180), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=46269]}, remoteAddress=/<IP_ADDRESS>:53926)
2022-07-21T15:48:50,058+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,058+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,063+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessagePublishContext - Success write topic: persistent://public/default/test-02-partition-0, producerName KOP-PID-PREFIX--1--1 ledgerId: 469, entryId: 8735 And triggered send callback.
2022-07-21T15:48:50,063+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.storage.ReplicaManager - Complete handle appendRecords.
2022-07-21T15:48:50,063+0000 [send-response-OrderedScheduler-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=126180), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=46269]}, remoteAddress=/<IP_ADDRESS>:53926) responseAndRequest content: {responses=[{topic=test-02,partition_responses=[{partition=0,error_code=0,base_offset=15670209,log_append_time=-1,log_start_offset=-1}]}],throttle_time_ms=0}
2022-07-21T15:48:50,087+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:57620] Received kafka cmd RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=152571), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=152571), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=8040]}, remoteAddress=/<IP_ADDRESS>:57620)
2022-07-21T15:48:50,087+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,087+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,091+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:43228] Received kafka cmd RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=135832), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=135832), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=6818]}, remoteAddress=/<IP_ADDRESS>:43228)
2022-07-21T15:48:50,091+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,091+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,091+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessagePublishContext - Success write topic: persistent://public/default/test-02-partition-0, producerName KOP-PID-PREFIX--1--1 ledgerId: 469, entryId: 8736 And triggered send callback.
2022-07-21T15:48:50,091+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.storage.ReplicaManager - Complete handle appendRecords.
2022-07-21T15:48:50,091+0000 [send-response-OrderedScheduler-3-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=152571), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=8040]}, remoteAddress=/<IP_ADDRESS>:57620) responseAndRequest content: {responses=[{topic=test-02,partition_responses=[{partition=0,error_code=0,base_offset=15670231,log_append_time=-1,log_start_offset=-1}]}],throttle_time_ms=0}
2022-07-21T15:48:50,095+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessagePublishContext - Success write topic: persistent://public/default/test-02-partition-0, producerName KOP-PID-PREFIX--1--1 ledgerId: 469, entryId: 8737 And triggered send callback.
2022-07-21T15:48:50,095+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.storage.ReplicaManager - Complete handle appendRecords.
2022-07-21T15:48:50,095+0000 [send-response-OrderedScheduler-2-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=135832), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=6818]}, remoteAddress=/<IP_ADDRESS>:43228) responseAndRequest content: {responses=[{topic=test-02,partition_responses=[{partition=0,error_code=0,base_offset=15670234,log_append_time=-1,log_start_offset=-1}]}],throttle_time_ms=0}
2022-07-21T15:48:50,103+0000 [send-response-OrderedScheduler-1-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7370), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670208,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254) responseAndRequest content: {throttle_time_ms=0,error_code=0,session_id=0,responses=[{topic=test-02,partition_responses=[{partition_header={partition=0,error_code=0,high_watermark=-1,last_stable_offset=-1,log_start_offset=-1,aborted_transactions=null},record_set=[]}]}]}
2022-07-21T15:48:50,103+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:55254] Received kafka cmd RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7371), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7371), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670208,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254)
2022-07-21T15:48:50,103+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Request RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7371) Fetch request. Size: 1. Each item:
2022-07-21T15:48:50,103+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler -   Fetch request topic:test-02-0 data:(offset=15670208, logStartOffset=-1, maxBytes=1048576).
2022-07-21T15:48:50,103+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,103+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,103+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Fetch for test-02-0: remove tcm to get cursor for fetch offset: 15670208 .
2022-07-21T15:48:50,103+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Get cursor for offset: 15670208 in cache. cache size: 0
2022-07-21T15:48:50,103+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Mark delete success for position: 469:8737
2022-07-21T15:48:50,103+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Topic test-02-0 success read entry: ledgerId: 469, entryId: 8737, size: 6819, ConsumerManager original offset: 15670208, lastEntryPosition: 469:8737, nextOffset: 15670237
2022-07-21T15:48:50,103+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Add cursor back kop-consumer-cursor-persistent://public/default/test-02-partition-0-414-2987-a2bc5bcbd3 for offset: 15670237
2022-07-21T15:48:50,103+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,103+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Request RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7371): read 4 entries but only 4 entries are committed
2022-07-21T15:48:50,104+0000 [send-response-OrderedScheduler-1-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7371), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670208,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254) responseAndRequest content: {throttle_time_ms=0,error_code=0,session_id=0,responses=[{topic=test-02,partition_responses=[{partition_header={partition=0,error_code=0,high_watermark=15670237,last_stable_offset=15670237,log_start_offset=15670237,aborted_transactions=[]},record_set=[(record=DefaultRecord(offset=15670208, timestamp=1658418529773, key=0 bytes, value=3633 bytes)), (record=DefaultRecord(offset=15670209, timestamp=1658418529994, key=0 bytes, value=2030 bytes)), (record=DefaultRecord(offset=15670210, timestamp=1658418529994, key=0 bytes, value=2042 bytes)), (record=DefaultRecord(offset=15670211, timestamp=1658418529994, key=0 bytes, value=2048 bytes)), (record=DefaultRecord(offset=15670212, timestamp=1658418529994, key=0 bytes, value=2067 bytes)), (record=DefaultRecord(offset=15670213, timestamp=1658418529994, key=0 bytes, value=2065 bytes)), (record=DefaultRecord(offset=15670214, timestamp=1658418529994, key=0 bytes, value=2035 bytes)), (record=DefaultRecord(offset=15670215, timestamp=1658418529994, key=0 bytes, value=2033 bytes)), (record=DefaultRecord(offset=15670216, timestamp=1658418529994, key=0 bytes, value=2038 bytes)), (record=DefaultRecord(offset=15670217, timestamp=1658418529994, key=0 bytes, value=2115 bytes)), (record=DefaultRecord(offset=15670218, timestamp=1658418529994, key=0 bytes, value=2115 bytes)), (record=DefaultRecord(offset=15670219, timestamp=1658418529994, key=0 bytes, value=2115 bytes)), (record=DefaultRecord(offset=15670220, timestamp=1658418529994, key=0 bytes, value=2114 bytes)), (record=DefaultRecord(offset=15670221, timestamp=1658418529994, key=0 bytes, value=2115 bytes)), (record=DefaultRecord(offset=15670222, timestamp=1658418529994, key=0 bytes, value=2115 bytes)), (record=DefaultRecord(offset=15670223, timestamp=1658418529994, key=0 bytes, value=2114 bytes)), (record=DefaultRecord(offset=15670224, timestamp=1658418529994, key=0 bytes, value=2114 bytes)), (record=DefaultRecord(offset=15670225, timestamp=1658418529994, key=0 bytes, value=2115 bytes)), (record=DefaultRecord(offset=15670226, timestamp=1658418529994, key=0 bytes, value=2115 bytes)), (record=DefaultRecord(offset=15670227, timestamp=1658418529994, key=0 bytes, value=2074 bytes)), (record=DefaultRecord(offset=15670228, timestamp=1658418529994, key=0 bytes, value=2367 bytes)), (record=DefaultRecord(offset=15670229, timestamp=1658418529994, key=0 bytes, value=2031 bytes)), (record=DefaultRecord(offset=15670230, timestamp=1658418529994, key=0 bytes, value=2033 bytes)), (record=DefaultRecord(offset=15670231, timestamp=1658418530046, key=0 bytes, value=2338 bytes)), (record=DefaultRecord(offset=15670232, timestamp=1658418530046, key=0 bytes, value=2810 bytes)), (record=DefaultRecord(offset=15670233, timestamp=1658418530046, key=0 bytes, value=2804 bytes)), (record=DefaultRecord(offset=15670234, timestamp=1658418530041, key=0 bytes, value=2183 bytes)), (record=DefaultRecord(offset=15670235, timestamp=1658418530041, key=0 bytes, value=2319 bytes)), (record=DefaultRecord(offset=15670236, timestamp=1658418530041, key=0 bytes, value=2228 bytes))]}]}]}
2022-07-21T15:48:50,105+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:55254] Received kafka cmd RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7372), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7372), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670237,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254)
2022-07-21T15:48:50,105+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Request RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7372) Fetch request. Size: 1. Each item:
2022-07-21T15:48:50,105+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler -   Fetch request topic:test-02-0 data:(offset=15670237, logStartOffset=-1, maxBytes=1048576).
2022-07-21T15:48:50,105+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,105+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,105+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Fetch for test-02-0: remove tcm to get cursor for fetch offset: 15670237 .
2022-07-21T15:48:50,105+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Get cursor for offset: 15670237 in cache. cache size: 0
2022-07-21T15:48:50,105+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Add cursor back kop-consumer-cursor-persistent://public/default/test-02-partition-0-414-2987-a2bc5bcbd3 for offset: 15670237
2022-07-21T15:48:50,105+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,105+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Request RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7372): read 0 entries but only 0 entries are committed
2022-07-21T15:48:50,142+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:46926] Received kafka cmd RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=100944), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=100944), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=6179]}, remoteAddress=/<IP_ADDRESS>:46926)
2022-07-21T15:48:50,142+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,142+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,145+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessagePublishContext - Success write topic: persistent://public/default/test-02-partition-0, producerName KOP-PID-PREFIX--1--1 ledgerId: 469, entryId: 8738 And triggered send callback.
2022-07-21T15:48:50,146+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.storage.ReplicaManager - Complete handle appendRecords.
2022-07-21T15:48:50,146+0000 [send-response-OrderedScheduler-2-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=100944), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=6179]}, remoteAddress=/<IP_ADDRESS>:46926) responseAndRequest content: {responses=[{topic=test-02,partition_responses=[{partition=0,error_code=0,base_offset=15670237,log_append_time=-1,log_start_offset=-1}]}],throttle_time_ms=0}
2022-07-21T15:48:50,154+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:57620] Received kafka cmd RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=152572), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=152572), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=5693]}, remoteAddress=/<IP_ADDRESS>:57620)
2022-07-21T15:48:50,154+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,154+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,158+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessagePublishContext - Success write topic: persistent://public/default/test-02-partition-0, producerName KOP-PID-PREFIX--1--1 ledgerId: 469, entryId: 8739 And triggered send callback.
2022-07-21T15:48:50,158+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.storage.ReplicaManager - Complete handle appendRecords.
2022-07-21T15:48:50,158+0000 [send-response-OrderedScheduler-3-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=152572), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=5693]}, remoteAddress=/<IP_ADDRESS>:57620) responseAndRequest content: {responses=[{topic=test-02,partition_responses=[{partition=0,error_code=0,base_offset=15670239,log_append_time=-1,log_start_offset=-1}]}],throttle_time_ms=0}
2022-07-21T15:48:50,185+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:53848] Received kafka cmd RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=314459), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=314459), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=19384]}, remoteAddress=/<IP_ADDRESS>:53848)
2022-07-21T15:48:50,185+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,185+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,189+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessagePublishContext - Success write topic: persistent://public/default/test-02-partition-0, producerName KOP-PID-PREFIX--1--1 ledgerId: 469, entryId: 8740 And triggered send callback.
2022-07-21T15:48:50,189+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.storage.ReplicaManager - Complete handle appendRecords.
2022-07-21T15:48:50,189+0000 [send-response-OrderedScheduler-2-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=314459), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=19384]}, remoteAddress=/<IP_ADDRESS>:53848) responseAndRequest content: {responses=[{topic=test-02,partition_responses=[{partition=0,error_code=0,base_offset=15670241,log_append_time=-1,log_start_offset=-1}]}],throttle_time_ms=0}
2022-07-21T15:48:50,198+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:53848] Received kafka cmd RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=314460), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=314460), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=177039]}, remoteAddress=/<IP_ADDRESS>:53848)
2022-07-21T15:48:50,198+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,198+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,204+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessagePublishContext - Success write topic: persistent://public/default/test-02-partition-0, producerName KOP-PID-PREFIX--1--1 ledgerId: 469, entryId: 8741 And triggered send callback.
2022-07-21T15:48:50,204+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.storage.ReplicaManager - Complete handle appendRecords.
2022-07-21T15:48:50,204+0000 [send-response-OrderedScheduler-2-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=314460), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=177039]}, remoteAddress=/<IP_ADDRESS>:53848) responseAndRequest content: {responses=[{topic=test-02,partition_responses=[{partition=0,error_code=0,base_offset=15670248,log_append_time=-1,log_start_offset=-1}]}],throttle_time_ms=0}
2022-07-21T15:48:50,205+0000 [send-response-OrderedScheduler-1-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7372), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670237,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254) responseAndRequest content: {throttle_time_ms=0,error_code=0,session_id=0,responses=[{topic=test-02,partition_responses=[{partition_header={partition=0,error_code=0,high_watermark=-1,last_stable_offset=-1,log_start_offset=-1,aborted_transactions=null},record_set=[]}]}]}
2022-07-21T15:48:50,205+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:55254] Received kafka cmd RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7373), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7373), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670237,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254)
2022-07-21T15:48:50,205+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Request RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7373) Fetch request. Size: 1. Each item:
2022-07-21T15:48:50,205+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler -   Fetch request topic:test-02-0 data:(offset=15670237, logStartOffset=-1, maxBytes=1048576).
2022-07-21T15:48:50,205+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,205+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,205+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Fetch for test-02-0: remove tcm to get cursor for fetch offset: 15670237 .
2022-07-21T15:48:50,205+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Get cursor for offset: 15670237 in cache. cache size: 0
2022-07-21T15:48:50,205+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Mark delete success for position: 469:8741
2022-07-21T15:48:50,205+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Topic test-02-0 success read entry: ledgerId: 469, entryId: 8741, size: 177467, ConsumerManager original offset: 15670237, lastEntryPosition: 469:8741, nextOffset: 15670312
2022-07-21T15:48:50,205+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Add cursor back kop-consumer-cursor-persistent://public/default/test-02-partition-0-414-2987-a2bc5bcbd3 for offset: 15670312
2022-07-21T15:48:50,205+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,205+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Request RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7373): read 4 entries but only 4 entries are committed
2022-07-21T15:48:50,208+0000 [send-response-OrderedScheduler-1-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7373), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670237,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254) responseAndRequest content: {throttle_time_ms=0,error_code=0,session_id=0,responses=[{topic=test-02,partition_responses=[{partition_header={partition=0,error_code=0,high_watermark=15670312,last_stable_offset=15670312,log_start_offset=15670312,aborted_transactions=[]},record_set=[(record=DefaultRecord(offset=15670237, timestamp=1658418528648, key=0 bytes, value=3029 bytes)), (record=DefaultRecord(offset=15670238, timestamp=1658418528648, key=0 bytes, value=3070 bytes)), (record=DefaultRecord(offset=15670239, timestamp=1658418530119, key=0 bytes, value=2810 bytes)), (record=DefaultRecord(offset=15670240, timestamp=1658418530119, key=0 bytes, value=2804 bytes)), (record=DefaultRecord(offset=15670241, timestamp=1658418530172, key=0 bytes, value=2760 bytes)), (record=DefaultRecord(offset=15670242, timestamp=1658418530172, key=0 bytes, value=2745 bytes)), (record=DefaultRecord(offset=15670243, timestamp=1658418530172, key=0 bytes, value=2745 bytes)), (record=DefaultRecord(offset=15670244, timestamp=1658418530172, key=0 bytes, value=2760 bytes)), (record=DefaultRecord(offset=15670245, timestamp=1658418530172, key=0 bytes, value=2745 bytes)), (record=DefaultRecord(offset=15670246, timestamp=1658418530172, key=0 bytes, value=2760 bytes)), (record=DefaultRecord(offset=15670247, timestamp=1658418530172, key=0 bytes, value=2745 bytes)), (record=DefaultRecord(offset=15670248, timestamp=1658418530184, key=0 bytes, value=2832 bytes)), (record=DefaultRecord(offset=15670249, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670250, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670251, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670252, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670253, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670254, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670255, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670256, timestamp=1658418530184, key=0 bytes, value=2832 bytes)), (record=DefaultRecord(offset=15670257, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670258, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670259, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670260, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670261, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670262, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670263, timestamp=1658418530184, key=0 bytes, value=2832 bytes)), (record=DefaultRecord(offset=15670264, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670265, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670266, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670267, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670268, timestamp=1658418530184, key=0 bytes, value=2797 bytes)), (record=DefaultRecord(offset=15670269, timestamp=1658418530184, key=0 bytes, value=2762 bytes)), (record=DefaultRecord(offset=15670270, timestamp=1658418530184, key=0 bytes, value=2734 bytes)), (record=DefaultRecord(offset=15670271, timestamp=1658418530184, key=0 bytes, value=2776 bytes)), (record=DefaultRecord(offset=15670272, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670273, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670274, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670275, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670276, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670277, timestamp=1658418530184, key=0 bytes, value=2751 bytes)), (record=DefaultRecord(offset=15670278, timestamp=1658418530184, key=0 bytes, value=2747 bytes)), (record=DefaultRecord(offset=15670279, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670280, timestamp=1658418530184, key=0 bytes, value=2747 bytes)), (record=DefaultRecord(offset=15670281, timestamp=1658418530184, key=0 bytes, value=2598 bytes)), (record=DefaultRecord(offset=15670282, timestamp=1658418530184, key=0 bytes, value=2638 bytes)), (record=DefaultRecord(offset=15670283, timestamp=1658418530184, key=0 bytes, value=2718 bytes)), (record=DefaultRecord(offset=15670284, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670285, timestamp=1658418530184, key=0 bytes, value=2762 bytes)), (record=DefaultRecord(offset=15670286, timestamp=1658418530184, key=0 bytes, value=2733 bytes)), (record=DefaultRecord(offset=15670287, timestamp=1658418530184, key=0 bytes, value=2747 bytes)), (record=DefaultRecord(offset=15670288, timestamp=1658418530184, key=0 bytes, value=2776 bytes)), (record=DefaultRecord(offset=15670289, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670290, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670291, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670292, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670293, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670294, timestamp=1658418530184, key=0 bytes, value=2728 bytes)), (record=DefaultRecord(offset=15670295, timestamp=1658418530184, key=0 bytes, value=2731 bytes)), (record=DefaultRecord(offset=15670296, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670297, timestamp=1658418530184, key=0 bytes, value=2762 bytes)), (record=DefaultRecord(offset=15670298, timestamp=1658418530184, key=0 bytes, value=2733 bytes)), (record=DefaultRecord(offset=15670299, timestamp=1658418530184, key=0 bytes, value=2747 bytes)), (record=DefaultRecord(offset=15670300, timestamp=1658418530184, key=0 bytes, value=2776 bytes)), (record=DefaultRecord(offset=15670301, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670302, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670303, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670304, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670305, timestamp=1658418530184, key=0 bytes, value=2769 bytes)), (record=DefaultRecord(offset=15670306, timestamp=1658418530184, key=0 bytes, value=2728 bytes)), (record=DefaultRecord(offset=15670307, timestamp=1658418530184, key=0 bytes, value=2731 bytes)), (record=DefaultRecord(offset=15670308, timestamp=1658418530184, key=0 bytes, value=2748 bytes)), (record=DefaultRecord(offset=15670309, timestamp=1658418530184, key=0 bytes, value=2556 bytes)), (record=DefaultRecord(offset=15670310, timestamp=1658418530184, key=0 bytes, value=2435 bytes)), (record=DefaultRecord(offset=15670311, timestamp=1658418530184, key=0 bytes, value=2554 bytes))]}]}]}
2022-07-21T15:48:50,208+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Channel writability has changed to: false
2022-07-21T15:48:50,208+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Channel writability has changed to: false
2022-07-21T15:48:50,208+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Channel writability has changed to: true
2022-07-21T15:48:50,208+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Channel writability has changed to: true
2022-07-21T15:48:50,209+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:55254] Received kafka cmd RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7374), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7374), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670312,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254)
2022-07-21T15:48:50,209+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Request RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7374) Fetch request. Size: 1. Each item:
2022-07-21T15:48:50,209+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler -   Fetch request topic:test-02-0 data:(offset=15670312, logStartOffset=-1, maxBytes=1048576).
2022-07-21T15:48:50,209+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,209+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,209+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Fetch for test-02-0: remove tcm to get cursor for fetch offset: 15670312 .
2022-07-21T15:48:50,209+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Get cursor for offset: 15670312 in cache. cache size: 0
2022-07-21T15:48:50,209+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Add cursor back kop-consumer-cursor-persistent://public/default/test-02-partition-0-414-2987-a2bc5bcbd3 for offset: 15670312
2022-07-21T15:48:50,209+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,209+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Request RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7374): read 0 entries but only 0 entries are committed
2022-07-21T15:48:50,220+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:53848] Received kafka cmd RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=314461), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=314461), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=5614]}, remoteAddress=/<IP_ADDRESS>:53848)
2022-07-21T15:48:50,220+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,220+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,221+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:47578] Received kafka cmd RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=155675), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=155675), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=2646]}, remoteAddress=/<IP_ADDRESS>:47578)
2022-07-21T15:48:50,221+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,221+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,223+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessagePublishContext - Success write topic: persistent://public/default/test-02-partition-0, producerName KOP-PID-PREFIX--1--1 ledgerId: 469, entryId: 8742 And triggered send callback.
2022-07-21T15:48:50,223+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.storage.ReplicaManager - Complete handle appendRecords.
2022-07-21T15:48:50,223+0000 [send-response-OrderedScheduler-2-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=314461), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=5614]}, remoteAddress=/<IP_ADDRESS>:53848) responseAndRequest content: {responses=[{topic=test-02,partition_responses=[{partition=0,error_code=0,base_offset=15670312,log_append_time=-1,log_start_offset=-1}]}],throttle_time_ms=0}
2022-07-21T15:48:50,225+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessagePublishContext - Success write topic: persistent://public/default/test-02-partition-0, producerName KOP-PID-PREFIX--1--1 ledgerId: 469, entryId: 8743 And triggered send callback.
2022-07-21T15:48:50,225+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.storage.ReplicaManager - Complete handle appendRecords.
2022-07-21T15:48:50,225+0000 [send-response-OrderedScheduler-2-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=155675), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=2646]}, remoteAddress=/<IP_ADDRESS>:47578) responseAndRequest content: {responses=[{topic=test-02,partition_responses=[{partition=0,error_code=0,base_offset=15670314,log_append_time=-1,log_start_offset=-1}]}],throttle_time_ms=0}
2022-07-21T15:48:50,234+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:47298] Received kafka cmd RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=87706), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=87706), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=12678]}, remoteAddress=/<IP_ADDRESS>:47298)
2022-07-21T15:48:50,234+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,234+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,238+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessagePublishContext - Success write topic: persistent://public/default/test-02-partition-0, producerName KOP-PID-PREFIX--1--1 ledgerId: 469, entryId: 8744 And triggered send callback.
2022-07-21T15:48:50,238+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.storage.ReplicaManager - Complete handle appendRecords.
2022-07-21T15:48:50,238+0000 [send-response-OrderedScheduler-1-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=87706), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=12678]}, remoteAddress=/<IP_ADDRESS>:47298) responseAndRequest content: {responses=[{topic=test-02,partition_responses=[{partition=0,error_code=0,base_offset=15670315,log_append_time=-1,log_start_offset=-1}]}],throttle_time_ms=0}
2022-07-21T15:48:50,274+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:44070] Received kafka cmd RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=166305), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=166305), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=2891]}, remoteAddress=/<IP_ADDRESS>:44070)
2022-07-21T15:48:50,274+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,274+0000 [pulsar-ph-kafka-64-1] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,278+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessagePublishContext - Success write topic: persistent://public/default/test-02-partition-0, producerName KOP-PID-PREFIX--1--1 ledgerId: 469, entryId: 8745 And triggered send callback.
2022-07-21T15:48:50,278+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.storage.ReplicaManager - Complete handle appendRecords.
2022-07-21T15:48:50,278+0000 [send-response-OrderedScheduler-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=PRODUCE, apiVersion=6, clientId=rdkafka, correlationId=166305), request={acks=-1,timeout=30000,partitionSizes=[test-02-0=2891]}, remoteAddress=/<IP_ADDRESS>:44070) responseAndRequest content: {responses=[{topic=test-02,partition_responses=[{partition=0,error_code=0,base_offset=15670318,log_append_time=-1,log_start_offset=-1}]}],throttle_time_ms=0}
2022-07-21T15:48:50,309+0000 [send-response-OrderedScheduler-1-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7374), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670312,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254) responseAndRequest content: {throttle_time_ms=0,error_code=0,session_id=0,responses=[{topic=test-02,partition_responses=[{partition_header={partition=0,error_code=0,high_watermark=-1,last_stable_offset=-1,log_start_offset=-1,aborted_transactions=null},record_set=[]}]}]}
2022-07-21T15:48:50,309+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:55254] Received kafka cmd RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7375), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7375), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670312,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254)
2022-07-21T15:48:50,309+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Request RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7375) Fetch request. Size: 1. Each item:
2022-07-21T15:48:50,309+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler -   Fetch request topic:test-02-0 data:(offset=15670312, logStartOffset=-1, maxBytes=1048576).
2022-07-21T15:48:50,309+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,309+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,309+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Fetch for test-02-0: remove tcm to get cursor for fetch offset: 15670312 .
2022-07-21T15:48:50,309+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Get cursor for offset: 15670312 in cache. cache size: 0
2022-07-21T15:48:50,309+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Mark delete success for position: 469:8745
2022-07-21T15:48:50,309+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Topic test-02-0 success read entry: ledgerId: 469, entryId: 8745, size: 2878, ConsumerManager original offset: 15670312, lastEntryPosition: 469:8745, nextOffset: 15670319
2022-07-21T15:48:50,309+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Add cursor back kop-consumer-cursor-persistent://public/default/test-02-partition-0-414-2987-a2bc5bcbd3 for offset: 15670319
2022-07-21T15:48:50,309+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,309+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Request RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7375): read 4 entries but only 4 entries are committed
2022-07-21T15:48:50,310+0000 [send-response-OrderedScheduler-1-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7375), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670312,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254) responseAndRequest content: {throttle_time_ms=0,error_code=0,session_id=0,responses=[{topic=test-02,partition_responses=[{partition_header={partition=0,error_code=0,high_watermark=15670319,last_stable_offset=15670319,log_start_offset=15670319,aborted_transactions=[]},record_set=[(record=DefaultRecord(offset=15670312, timestamp=1658418530214, key=0 bytes, value=2668 bytes)), (record=DefaultRecord(offset=15670313, timestamp=1658418530214, key=0 bytes, value=2867 bytes)), (record=DefaultRecord(offset=15670314, timestamp=1658418530216, key=0 bytes, value=2576 bytes)), (record=DefaultRecord(offset=15670315, timestamp=1658418521784, key=0 bytes, value=4222 bytes)), (record=DefaultRecord(offset=15670316, timestamp=1658418521784, key=0 bytes, value=4220 bytes)), (record=DefaultRecord(offset=15670317, timestamp=1658418521784, key=0 bytes, value=4146 bytes)), (record=DefaultRecord(offset=15670318, timestamp=1658418530169, key=0 bytes, value=2821 bytes))]}]}]}
2022-07-21T15:48:50,310+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:55254] Received kafka cmd RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7376), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7376), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670319,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254)
2022-07-21T15:48:50,310+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Request RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7376) Fetch request. Size: 1. Each item:
2022-07-21T15:48:50,310+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler -   Fetch request topic:test-02-0 data:(offset=15670319, logStartOffset=-1, maxBytes=1048576).
2022-07-21T15:48:50,310+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,310+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,310+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Fetch for test-02-0: remove tcm to get cursor for fetch offset: 15670319 .
2022-07-21T15:48:50,310+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Get cursor for offset: 15670319 in cache. cache size: 0
2022-07-21T15:48:50,310+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Add cursor back kop-consumer-cursor-persistent://public/default/test-02-partition-0-414-2987-a2bc5bcbd3 for offset: 15670319
2022-07-21T15:48:50,310+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,310+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Request RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7376): read 0 entries but only 0 entries are committed
2022-07-21T15:48:50,410+0000 [send-response-OrderedScheduler-1-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - Write kafka cmd to client. request content: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7376), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670319,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254) responseAndRequest content: {throttle_time_ms=0,error_code=0,session_id=0,responses=[{topic=test-02,partition_responses=[{partition_header={partition=0,error_code=0,high_watermark=-1,last_stable_offset=-1,log_start_offset=-1,aborted_transactions=null},record_set=[]}]}]}
2022-07-21T15:48:50,411+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaCommandDecoder - [/<IP_ADDRESS>:55254] Received kafka cmd RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7377), the request content is: KafkaHeaderAndRequest(header=RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7377), request={replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=52428800,isolation_level=1,session_id=0,epoch=-1,topics=[{topic=test-02,partitions=[{partition=0,fetch_offset=15670319,log_start_offset=-1,max_bytes=1048576}]}],forgotten_topics_data=[]}, remoteAddress=/<IP_ADDRESS>:55254)
2022-07-21T15:48:50,411+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Request RequestHeader(apiKey=FETCH, apiVersion=8, clientId=vector, correlationId=7377) Fetch request. Size: 1. Each item:
2022-07-21T15:48:50,411+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler -   Fetch request topic:test-02-0 data:(offset=15670319, logStartOffset=-1, maxBytes=1048576).
2022-07-21T15:48:50,411+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,411+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaRequestHandler - using public as tenant
2022-07-21T15:48:50,411+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.MessageFetchContext - Fetch for test-02-0: remove tcm to get cursor for fetch offset: 15670319 .
2022-07-21T15:48:50,411+0000 [pulsar-ph-kafka-64-2] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Get cursor for offset: 15670319 in cache. cache size: 0
2022-07-21T15:48:50,411+0000 [BookKeeperClientWorker-OrderedExecutor-0-0] DEBUG io.streamnative.pulsar.handlers.kop.KafkaTopicConsumerManager - [[id: 0xf76a13f2, L:/<IP_ADDRESS>:9092 - R:/<IP_ADDRESS>:55254]] Add cursor back kop-consumer-cursor-persistent://public/default/test-02-partition-0-414-2987-a2bc5bcbd3 for offset: 15670319

It looks okay from the logs. What's the appearance? BTW, the CoordinatorLoadInProgress (Broker: Coordinator load in progress) is not a permanent error, it's just because the group coordinator (for consumer) or transaction coordinator (for producer) is loading the metadata from the storage topic (__consumer_offsets or __transaction_state).

broker.conf.zip

@jiangtao7 I cannot find the transaction related configs in your broker.conf.
brokerDeduplicationEnabled=true
kafkaTransactionCoordinatorEnabled=true

Did you miss the comments here: https://github.com/streamnative/kop/issues/1415#issuecomment-1191002775?

I also see you configured the same advertised listener and listener
kafkaAdvertisedListeners=PLAINTEXT://pulsar-broker-0.pulsar-broker.pulsar-poc.svc.cluster.local:9092
kafkaListeners=PLAINTEXT://pulsar-broker-0.pulsar-broker.pulsar-poc.svc.cluster.local:9092

The kafkaAdvertisedListeners should be like the advertised.listeners while the kafkaListeners should be like the listeners config in Kafka. It means you should configure an external address for kafkaAdvertisedListeners  and an internal address for kafkaListeners.
See https://cwiki.apache.org/confluence/display/KAFKA/KIP-103%3A+Separation+of+Internal+and+External+traffic.

We did set to above, attaching the entire manifest and values.yaml
`

Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements.  See the NOTICE file
distributed with this work for additional information
regarding copyright ownership.  The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License.  You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied.  See the License for the
specific language governing permissions and limitations
under the License.

Flag to control whether to run initialize job
initialize: true

K8S Settings

Namespace to deploy pulsar
NOTE: Make the default namespace as empty. So it will fallback to use the namespace used for installing the helm
chart. Helm does not position it self as a namespace manager, as namespaces in kubernetes are considered as
a higher control structure that is not part of the application.
namespace: "pulsar-poc"
namespaceCreate: true

Global Settings

Pulsar Metadata Prefix

By default, pulsar stores all the metadata at root path.
You can configure to have a prefix (e.g. "/my-pulsar-cluster").
If you do so, all the pulsar and bookkeeper metadata will
be stored under the provided path
metadataPrefix: ""
Persistence

If persistence is enabled, components that have state will
be deployed with PersistentVolumeClaims, otherwise, for test
purposes, they will be deployed with emptyDir

This is a global setting that is applied to all components.
If you need to disable persistence for a component,
you can set the volume.persistence setting to false for
that component.
volumes:
persistence: true
configure the components to use local persistent volume
the local provisioner should be installed prior to enable local persistent volume
local_storage: false
AntiAffinity

Flag to enable and disable AntiAffinity for all components.
This is a global setting that is applied to all components.
If you need to disable AntiAffinity for a component, you can set
the affinity.anti_affinity settings to false for that component.
affinity:
anti_affinity: true
Deployment type

Control what configurations of Apache Pulsar to deploy for a specified environment type
deployment:
This option will configure specific resources/object for a deployment on openshift
openshift: false
Components

Control what components of Apache Pulsar to deploy for the cluster
components:
zookeeper
zookeeper: true
bookkeeper
bookkeeper: true
bookkeeper - autorecovery
autorecovery: true
broker
broker: true
functions
functions: true
proxy
proxy: true
toolset
toolset: true
pulsar manager
pulsar_manager: true
pulsar sql
sql_worker: false
kop
kop: true
mop
mop: false
pulsar detector
pulsar_detector: true
superset
superset: false
Monitoring Components

Control what components of the monitoring stack to deploy for the cluster
monitoring:
monitoring - prometheus
prometheus: false
monitoring - grafana
grafana: false
monitoring - node_exporter
node_exporter: false
alerting - alert-manager
alert_manager: false
monitoring - loki
loki: false
monitoring - datadog
datadog: false
Images

Control what images to use for each component
images:
zookeeper:
repository: streamnative/pulsar-all
tag: <IP_ADDRESS>
pullPolicy: IfNotPresent
bookie:
repository: streamnative/pulsar-all
tag: <IP_ADDRESS>
pullPolicy: IfNotPresent
presto:
repository: streamnative/pulsar-all
tag: <IP_ADDRESS>
pullPolicy: IfNotPresent
NOTE: allow overriding the presto worker image
presto_worker:
repository: streamnative/pulsar-all
tag: <IP_ADDRESS>
pullPolicy: IfNotPresent
autorecovery:
repository: streamnative/pulsar-all
tag: <IP_ADDRESS>
pullPolicy: IfNotPresent
broker:
repository: streamnative/sn-pulsar
tag: <IP_ADDRESS>
pullPolicy: IfNotPresent
proxy:
repository: streamnative/sn-pulsar
tag: <IP_ADDRESS>
pullPolicy: IfNotPresent
pulsar_detector:
repository: streamnative/sn-platform
tag: <IP_ADDRESS>
pullPolicy: IfNotPresent
functions:
repository: streamnative/sn-pulsar
tag: <IP_ADDRESS>
pullPolicy: IfNotPresent
function_worker:
repository: streamnative/sn-pulsar
tag: <IP_ADDRESS>
pullPolicy: IfNotPresent
function_runner_java:
repository: streamnative/pulsar-functions-java-runner
tag: <IP_ADDRESS>
pullPolicy: IfNotPresent
function_runner_go:
repository: streamnative/pulsar-functions-go-runner
tag: <IP_ADDRESS>
pullPolicy: IfNotPresent
function_runner_python:
repository: streamnative/pulsar-functions-python-runner
tag: <IP_ADDRESS>
pullPolicy: IfNotPresent
NOTE: allow overriding the toolset image
toolset:
repository: streamnative/pulsar-all
tag: <IP_ADDRESS>
pullPolicy: IfNotPresent
prometheus:
repository: prom/prometheus
tag: v2.17.2
pullPolicy: IfNotPresent
alert_manager:
repository: prom/alertmanager
tag: v0.20.0
pullPolicy: IfNotPresent
grafana:
repository: streamnative/apache-pulsar-grafana-dashboard-k8s
tag: 0.0.17
pullPolicy: IfNotPresent
pulsar_manager:
repository: streamnative/pulsar-manager
tag: 0.3.0
pullPolicy: IfNotPresent
hasCommand: false
node_exporter:
repository: prom/node-exporter
tag: v0.16.0
pullPolicy: "IfNotPresent"
nginx_ingress_controller:
repository: quay.io/kubernetes-ingress-controller/nginx-ingress-controller
tag: 0.26.2
pullPolicy: "IfNotPresent"
TLS
templates/tls-certs.yaml

The chart is using cert-manager for provisioning TLS certs for
brokers and proxies.
tls:
enabled: false
common settings for generating certs
common:
# 10yrs
duration: 87660h
# 15d
renewBefore: 360h
organizations: [pulsar]
privateKey:
size: 4096
algorithm: RSA
encoding: PKCS8
caSecretName:
settings for generating certs for proxy
proxy:
enabled: false
cert_name: tls-proxy
certSecretName:
untrustedCa: true
settings for generating certs for proxy
pulsar_detector:
enabled: false
cert_name: tls-pulsar-detector
settings for generating certs for broker
broker:
enabled: false
cert_name: tls-broker
certSecretName:
brokerClient:
enabled: false
functions:
enabled: false
cert_name: tls-function
certSecretName:
settings for generating certs for kop
kop:
enabled: false
jks:
password: "[pulsar tls jks store password]"
cert_name: tls-kop
settings for generating certs for bookies
bookie:
enabled: false
cert_name: tls-bookie
certSecretName:
settings for generating certs for zookeeper
zookeeper:
enabled: false
cert_name: tls-zookeeper
certSecretName:
settings for generating certs for recovery
autorecovery:
cert_name: tls-recovery
certSecretName:
settings for generating certs for toolset
toolset:
cert_name: tls-toolset
certSecretName:
pulsar_manager:
enabled: false
cert_name: tls-pulsar-manager
certSecretName:
presto:
enabled: false
cert_name: tls-presto
certSecretName:
Enable or disable broker authentication and authorization.
auth:
authentication:
enabled: false
provider: "jwt"
jwt:
# Enable JWT authentication
# If the token is generated by a secret key, set the usingSecretKey as true.
# If the token is generated by a private key, set the usingSecretKey as false.
usingSecretKey: false
authorization:
enabled: false
superUsers:
# broker to broker communication
broker: "broker-admin"
# proxy to broker communication
proxy: "proxy-admin"
# websocket proxy to broker communication
websocket: "ws-admin"
# pulsar-admin client to broker/proxy communication
client: "admin"
# pulsar-manager to broker/proxy communication
pulsar_manager: "pulsar-manager-admin"
Enable vault based authentication
vault:
enabled: false
######################################################################
External dependencies
######################################################################
cert-manager
templates/tls-cert-issuer.yaml

Cert manager is used for automatically provisioning TLS certificates
for components within a Pulsar cluster
certs:
internal_issuer:
enabled: true
component: internal-cert-issuer
type: selfsigning
public_issuer:
enabled: true
component: public-cert-issuer
type: acme
api_version: "cert-manager.io/v1"
issuers:
selfsigning:
acme:
# You must replace this email address with your own.
# Let's Encrypt will use this to contact you about expiring
# certificates, and issues related to your account.
email<EMAIL_ADDRESS># change this to production endpoint once you successfully test it
# server: https://acme-v02.api.letsencrypt.org/directory
server: https://acme-staging-v02.api.letsencrypt.org/directory
solver: clouddns
solvers:
clouddns:
# TODO: add a link about how to configure this section
project: "[YOUR GCP PROJECT ID]"
serviceAccountSecretRef:
name: "[NAME OF SECRET]"
key: "[KEY OF SECRET]"
# route53:
#   region: "[ROUTE53 REGION]"
#   secretAccessKeySecretRef:
#     name: "[NAME OF SECRET]"
#     key: "[KEY OF SECRET]"
#   role: "[ASSUME A ROLE]"
lets_encrypt:
ca_ref:
secretName: "pulsar-tls-proxy"
keyName: "ca.crt"
External DNS
templates/external-dns.yaml
templates/external-dns-rbac.yaml

External DNS is used for synchronizing exposed Ingresses with DNS providers
external_dns:
enabled: false
use_existing: false
component: external-dns
policy: upsert-only
registry: txt
owner_id: pulsar
domain_filter: pulsar.example.local
provider: google
providers:
google:
# project: external-dns-test
project: "[GOOGLE PROJECT ID]"
aws:
zoneType: public
serviceAcct:
annotations: {}
securityContext: {}
extraMounts: []
extraEnv: []
Domain requested from External DNS
domain:
enabled: false
suffix: test.pulsar.example.local
Ingresses for exposing Pulsar services
ingress:
templates/proxy-service-ingress.yaml

Ingresses for exposing pulsar service publicly
proxy:
enabled: false
tls:
enabled: true
plaintext:
enabled: false
# tls.enabled will turn on all the TLS ports
# plaintext.enabled will control whether to turn on all the plaintext ports
# this section is used to control individal ports
ports:
http: true
https: true
pulsar: true
pulsarssl: true
websocket: true
websockettls: true
portNumbers:
https: 443
type: LoadBalancer
annotations: {}
extraSpec: {}
templates/broker-service-ingress.yaml

Ingresses for exposing pulsar service publicly
broker:
enabled: false
type: LoadBalancer
annotations: {}
extraSpec: {}
kop:
enabled: false
# NOTE: replace this with the actual domain when enabling ingress for KOP
external_domain: "eks01.uw2.aws-omnicloud.com"
templates/presto-service-ingress.yaml

Ingresses for exposing presto service publicly
presto:
enabled: false
tls:
enabled: true
type: LoadBalancer
annotations: {}
extraSpec: {}
ports:
http: 80
https: 443
templates/control-center-ingress.yaml

Ingresses for exposing monitoring/management services publicly
controller:
enabled: false
rbac: true
component: nginx-ingress-controller
replicaCount: 1
# nodeSelector:
# cloud.google.com/gke-nodepool: default-pool
tolerations: []
gracePeriod: 300
annotations: {}
ports:
http: 80
https: 443
# flag whether to terminate the tls at the loadbalancer level
tls:
termination: false
control_center:
enabled: true
component: control-center
endpoints:
grafana: true
prometheus: false
alertmanager: false
# Set external domain of the load balancer of ingress controller
# external_domain: your.external.control.center.domain
# external_domain_scheme: https://
tls:
enabled: false
annotations: {}
imagePuller:
component: image-puller
pullSecret:
enabled: false
hook:
enabled: false
image:
name: streamnative/k8s-image-awaiter
tag: '0.1.0'
rbac:
enabled: true
continuous:
enabled: false
pause:
image:
name: gcr.io/google_containers/pause
tag: '3.1'
imagePullSecrets: regcred
######################################################################
Below are settings for each component
######################################################################
Common properties applied to pulsar components
common:
extraInitContainers: []
Pulsar: Zookeeper cluster
templates/zookeeper-statefulset.yaml

zookeeper:
use a component name that matches your grafana configuration
so the metrics are correctly rendered in grafana dashboard
component: zookeeper
the number of zookeeper servers to run. it should be an odd number larger than or equal to 3.
replicaCount: 3
updateStrategy: RollingUpdate
ports:
metrics: 8000
client: 2181
clientTls: 2281
follower: 2888
leaderElection: 3888
nodeSelector:
# cloud.google.com/gke-nodepool: default-pool

probe:
liveness:
enabled: true
failureThreshold: 10
initialDelaySeconds: 10
periodSeconds: 30
timeoutSeconds: 5
readiness:
enabled: true
failureThreshold: 10
initialDelaySeconds: 10
periodSeconds: 30
timeoutSeconds: 5
startup:
enabled: false
failureThreshold: 30
initialDelaySeconds: 10
periodSeconds: 30
affinity:
anti_affinity: true
# Set the anti affinity type. Valid values:
# requiredDuringSchedulingIgnoredDuringExecution - rules must be met for pod to be scheduled (hard) requires at least one node per replica
# preferredDuringSchedulingIgnoredDuringExecution - scheduler will try to enforce but not guranentee
type: preferredDuringSchedulingIgnoredDuringExecution
annotations:
prometheus.io/scrape: "true"
prometheus.io/port: "8000"
securityContext:
fsGroup: 0
fsGroupChangePolicy: "OnRootMismatch"
tolerations: []
gracePeriod: 30
resources:
requests:
memory: 256Mi
cpu: 0.1
volumes:
# use a persistent volume or emptyDir
persistence: true
# Add a flag here for backward compatibility. Ideally we should
# use two disks for production workloads. This flag might be
# removed in the future releases to stick to two-disks mode.
useSeparateDiskForTxlog: false
data:
name: data
size: 50Gi
local_storage: true
# storageClassName: ""
## If the storage class is left undefined when using persistence
## the default storage class for the cluster will be used.
##
# storageClass:
#   type: pd-ssd
#   fsType: xfs
#   provisioner: kubernetes.io/gce-pd
#   allowVolumeExpansion: false
#   volumeBindingMode: Immediate
#   reclaimPolicy: Retain
#   allowedTopologies:
#   mountOptions:
#   extraParameters:
#     iopsPerGB: "50"
dataLog:
name: datalog
size: 10Gi
local_storage: true
# storageClassName: ""
## If the storage class is left undefined when using persistence
## the default storage class for the cluster will be used.
##
# storageClass:
#   type: pd-ssd
#   fsType: xfs
#   provisioner: kubernetes.io/gce-pd
#   allowVolumeExpansion: false
#   volumeBindingMode: Immediate
#   reclaimPolicy: Retain
#   allowedTopologies:
#   mountOptions:
#   extraParameters:
#     iopsPerGB: "50"
extraInitContainers: []
Zookeeper configmap
templates/zookeeper-configmap.yaml

The initial myid used for generating myid for each zookeeper pod.
initialMyId: 0
peerType: "participant"
reconfig settings
reconfig:
enabled: false
# The zookeeper servers to observe/join
zkServers: []
Automtically Roll Deployments when configmap is changed
autoRollDeployment: true
useNettyIO: true
configData:
PULSAR_MEM: >
-Xms64m -Xmx128m
PULSAR_GC: >
-XX:+UseG1GC
-XX:MaxGCPauseMillis=10
-Dcom.sun.management.jmxremote
-Djute.maxbuffer=10485760
-XX:+ParallelRefProcEnabled
-XX:+UnlockExperimentalVMOptions
-XX:+AggressiveOpts
-XX:+DoEscapeAnalysis
-XX:+DisableExplicitGC
-XX:+PerfDisableSharedMem
-Dzookeeper.forceSync=no
Zookeeper service
templates/zookeeper-service.yaml

service:
annotations:
service.alpha.kubernetes.io/tolerate-unready-endpoints: "true"
Zookeeper PodDisruptionBudget
templates/zookeeper-pdb.yaml

pdb:
usePolicy: true
maxUnavailable: 1
customTools:
serviceAccount:
use: true
create: true
name: ""
annotations: {}
clusterRole: true
backup:
component: "backup"
enable: false
repository: "streamnative/pulsar-metadata-tool"
tag: "<IP_ADDRESS>"
pullPolicy: IfNotPresent
webServerPort: "8088"
backupInterval: "600"
bucket: "s3a://bucket"
backupPrefix: "pulsar-backup"
managedLedgerPath: "/managed-ledgers"
configData:
# extra opts for the java command
OPTS: ""
secrets:
use: false
#    aws:
#      secretName: "backup-aws-secret"
## secret that stores AWS credentials. The secret should be created in the following
## format.
##     ## kubectl -n pulsar create secret generic \     ##   --from-literal=AWS_ACCESS_KEY_ID=[AWS ACCESS KEY] \     ##   --from-literal=AWS_SECRET_ACCESS_KEY=[AWS SECRET KEY] \     ##   [secret name]     ##
restore:
component: "restore"
enable: false
repository: "streamnative/pulsar-metadata-tool"
tag: "<IP_ADDRESS>"
pullPolicy: IfNotPresent
restorePoint: ""
restoreVersion: "1"
managedLedgerPath: "/managed-ledgers"
bucket: "s3a://bucket"
cleanupBookieMeta: false
cleanupClusterMeta: false
configData:
OPTS: ""
secrets:
use: false
aws:
secretName: "restore-aws-secret"
Pulsar: Bookkeeper cluster
templates/bookkeeper-statefulset.yaml

bookkeeper:
use a component name that matches your grafana configuration
so the metrics are correctly rendered in grafana dashboard
component: bookie
BookKeeper Cluster Initialize
templates/bookkeeper-cluster-initialize.yaml
metadata:
## Set the resources used for running bin/bookkeeper shell initnewcluster
##
resources:
# requests:
# memory: 4Gi
# cpu: 2
replicaCount: 4
ports:
http: 8000
bookie: 3181
streamStorage: 4181
nodeSelector:
# cloud.google.com/gke-nodepool: default-pool

probe:
liveness:
enabled: true
failureThreshold: 60
initialDelaySeconds: 10
periodSeconds: 30
readiness:
enabled: true
failureThreshold: 60
initialDelaySeconds: 10
periodSeconds: 30
startup:
enabled: false
failureThreshold: 30
initialDelaySeconds: 60
periodSeconds: 30
affinity:
anti_affinity: true
# Set the anti affinity type. Valid values:
# requiredDuringSchedulingIgnoredDuringExecution - rules must be met for pod to be scheduled (hard) requires at least one node per replica
# preferredDuringSchedulingIgnoredDuringExecution - scheduler will try to enforce but not guranentee
type: preferredDuringSchedulingIgnoredDuringExecution
annotations: {}
securityContext:
fsGroup: 0
fsGroupChangePolicy: "OnRootMismatch"
tolerations: []
gracePeriod: 30
resources:
requests:
memory: 512Mi
cpu: 0.2
Definition of the serviceAccount used to run bookies.
serviceAccount:
# Specifies whether to use a service account to run this component
use: true
# Specifies whether a service account should be created
create: true
# The name of the service account to use.
# If not set and create is true, a name is generated using the fullname template
name: ""
# whether to create a cluster role
clusterRole: true
# Extra annotations for the serviceAccount definition. This can either be
# YAML or a YAML-formatted multi-line templated string map of the
# annotations to apply to the serviceAccount.
annotations: {}
volumes:
# use a persistent volume or emptyDir
persistence: true
journal:
name: journal
size: 10Gi
local_storage: true
# storageClassName: ""
## If the storage class is left undefined when using persistence
## the default storage class for the cluster will be used.
##
# storageClass:
#   type: pd-ssd
#   fsType: xfs
#   provisioner: kubernetes.io/gce-pd
#   allowVolumeExpansion: false
#   volumeBindingMode: Immediate
#   reclaimPolicy: Retain
#   allowedTopologies:
#   mountOptions:
#   extraParameters:
#     iopsPerGB: "50"
ledgers:
name: ledgers
size: 50Gi
local_storage: true
# storageClassName: ""
## If the storage class is left undefined when using persistence
## the default storage class for the cluster will be used.
##
# storageClass:
# type: pd-ssd
# fsType: xfs
# provisioner: kubernetes.io/gce-pd
# allowVolumeExpansion: false
# volumeBindingMode: Immediate
# reclaimPolicy: Retain
# allowedTopologies:
# mountOptions:
# extraParameters:
#   iopsPerGB: "50"
extraInitContainers: []
Bookkeeper configmap
templates/bookkeeper-configmap.yaml

Automtically Roll Deployments when configmap is changed
autoRollDeployment: true
placementPolicy:
rackAware: true
configData:
# BOOKIE_MEM is used for bookie shell
BOOKIE_MEM: >
-Xms128m
-Xmx256m
-XX:MaxDirectMemorySize=256m
# we use bin/pulsar for starting bookie daemons
PULSAR_MEM: >
-Xms128m
-Xmx256m
-XX:MaxDirectMemorySize=256m
PULSAR_GC: >
-XX:+UseG1GC
-XX:MaxGCPauseMillis=10
-XX:+ParallelRefProcEnabled
-XX:+UnlockExperimentalVMOptions
-XX:+AggressiveOpts
-XX:+DoEscapeAnalysis
-XX:ParallelGCThreads=4
-XX:ConcGCThreads=4
-XX:G1NewSizePercent=50
-XX:+DisableExplicitGC
-XX:-ResizePLAB
-XX:+ExitOnOutOfMemoryError
-XX:+PerfDisableSharedMem
-verbosegc
Bookkeeper Service
templates/bookkeeper-service.yaml

service:
annotations:
publishNotReadyAddresses: "true"
Bookkeeper PodDisruptionBudget
templates/bookkeeper-pdb.yaml

pdb:
usePolicy: true
maxUnavailable: 1
Pulsar: Bookkeeper AutoRecovery
templates/autorecovery-statefulset.yaml

autorecovery:
use a component name that matches your grafana configuration
so the metrics are correctly rendered in grafana dashboard
component: recovery
replicaCount: 1
ports:
http: 8000
nodeSelector:
# cloud.google.com/gke-nodepool: default-pool

affinity:
anti_affinity: true
# Set the anti affinity type. Valid values:
# requiredDuringSchedulingIgnoredDuringExecution - rules must be met for pod to be scheduled (hard) requires at least one node per replica
# preferredDuringSchedulingIgnoredDuringExecution - scheduler will try to enforce but not guranentee
type: preferredDuringSchedulingIgnoredDuringExecution
annotations: {}
securityContext: {}
tolerations: []
gracePeriod: 30
resources:
requests:
memory: 64Mi
cpu: 0.05
extraInitContainers: []
Bookkeeper auto-recovery configmap
templates/autorecovery-configmap.yaml

Automtically Roll Deployments when configmap is changed
autoRollDeployment: true
configData:
BOOKIE_MEM: >
-Xms64m -Xmx64m
Pulsar Zookeeper metadata. The metadata will be deployed as
soon as the last zookeeper node is reachable. The deployment
of other components that depends on zookeeper, such as the
bookkeeper nodes, broker nodes, etc will only start to be
deployed when the zookeeper cluster is ready and with the
metadata deployed
pulsar_metadata:
component: pulsar-init
image:
# the image used for running pulsar-cluster-initialize job
# repository: apachepulsar/pulsar-all
# tag: 2.5.0
repository: streamnative/pulsar-all
tag: <IP_ADDRESS>
pullPolicy: IfNotPresent
set an existing configuration store
configurationStore:
configurationStoreMetadataPrefix: ""
optional, you can provide your own zookeeper metadata store for other components
to use this, you should explicit set components.zookeeper to false

userProvidedZookeepers: "zk01.example.com:2181,zk02.example.com:2181"
set the cluster name. if empty or not specified,
it will use helm release name to generate a cluster name.
clusterName: ""
Pulsar: KoP Protocol Handler
kop:
ports:
plaintext: 9092
ssl: 9093
auth:
enabled: false
Pulsar: MoP Protocol Handler
mop:
ports:
plaintext: 1883
tls_psk: 5684
TLS PSK is introduced since sn-pulsar:<IP_ADDRESS>
tls_psk:
enabled: false
identity: "[ADD IDENTITY]"
identityHint: "[ADD HINT]"
secret: "[ADD SECRET]"
Istio is required for exposing Kafka service
istio:
enable: true
gateway:
selector:
istio: "ingress"
Pulsar: Broker cluster
templates/broker-statefulset.yaml

broker:
use a component name that matches your grafana configuration
so the metrics are correctly rendered in grafana dashboard
component: broker
replicaCount: 3
ports:
http: 8080
https: 8443
pulsar: 6650
pulsarssl: 6651
nodeSelector:
# cloud.google.com/gke-nodepool: default-pool

probe:
liveness:
enabled: true
failureThreshold: 10
initialDelaySeconds: 30
periodSeconds: 10
readiness:
enabled: true
failureThreshold: 10
initialDelaySeconds: 30
periodSeconds: 10
startup:
enabled: false
failureThreshold: 30
initialDelaySeconds: 60
periodSeconds: 10
affinity:
anti_affinity: true
# Set the anti affinity type. Valid values:
# requiredDuringSchedulingIgnoredDuringExecution - rules must be met for pod to be scheduled (hard) requires at least one node per replica
# preferredDuringSchedulingIgnoredDuringExecution - scheduler will try to enforce but not guranentee
type: preferredDuringSchedulingIgnoredDuringExecution
annotations:
sidecar.istio.io/inject: "true"
tolerations: []
securityContext: {}
gracePeriod: 30
flag to advertise pod ip address
advertisedPodIP: false
resources:
requests:
memory: 512Mi
cpu: 0.2
extraInitContainers: []
extraVolumes: []
extraVolumeMounts: []
extraEnv: []
Definition of the serviceAccount used to run brokers.
serviceAccount:
# Specifies whether to use a service account to run this component
use: true
# Specifies whether a service account should be created
create: true
# The name of the service account to use.
# If not set and create is true, a name is generated using the fullname template
name: ""
# Extra annotations for the serviceAccount definition. This can either be
# YAML or a YAML-formatted multi-line templated string map of the
# annotations to apply to the serviceAccount.
annotations: {}
# whether to create a cluster role
clusterRole: true
Broker configmap
templates/broker-configmap.yaml

Automtically Roll Deployments when configmap is changed
autoRollDeployment: true
configData:
PULSAR_MEM: >
-Xms128m -Xmx256m -XX:MaxDirectMemorySize=256m
PULSAR_GC: >
-XX:+UseG1GC
-XX:MaxGCPauseMillis=10
-Dio.netty.leakDetectionLevel=disabled
-Dio.netty.recycler.linkCapacity=1024
-XX:+ParallelRefProcEnabled
-XX:+UnlockExperimentalVMOptions
-XX:+AggressiveOpts
-XX:+DoEscapeAnalysis
-XX:ParallelGCThreads=4
-XX:ConcGCThreads=4
-XX:G1NewSizePercent=50
-XX:+DisableExplicitGC
-XX:-ResizePLAB
-XX:+ExitOnOutOfMemoryError
-XX:+PerfDisableSharedMem
# @deprecated:
# 1. You can use a service account to assume an IAM role to access S3 bucket
# 2. You can use a k8s secret to reference AWS credentials. See offload.s3.secret
# AWS_ACCESS_KEY_ID: "[YOUR AWS ACCESS KEY ID]"
# AWS_SECRET_ACCESS_KEY: "[YOUR SECRET]"
managedLedgerDefaultEnsembleSize: "3"
managedLedgerDefaultWriteQuorum: "3"
managedLedgerDefaultAckQuorum: "2"
messagingProtocols: "kafka"
protocolHandlerDirectory: "./protocols"
kafkaListeners: "PLAINTEXT://<IP_ADDRESS>:9092"
allowAutoTopicCreationType: "partitioned"
brokerDeleteInactiveTopicsEnabled: "false"
brokerDeduplicationEnabled: "true"
kafkaTransactionCoordinatorEnabled: "true"
Broker service
templates/broker-service.yaml

service:
annotations: {}
Broker PodDisruptionBudget
templates/broker-pdb.yaml

pdb:
usePolicy: true
maxUnavailable: 1
Broker service account
templates/broker-service-account.yaml
deprecated: use serviceAccount section to configure service account.
service_account:
annotations: {}
offload:
enabled: false
managedLedgerOffloadDriver: aws-s3
gcs:
enabled: false
gcsManagedLedgerOffloadRegion: "[YOUR REGION OF GCS]"
gcsManagedLedgerOffloadBucket: "[YOUR BUCKET OF GCS]"
gcsManagedLedgerOffloadMaxBlockSizeInBytes: "67108864"
gcsManagedLedgerOffloadReadBufferSizeInBytes: "1048576"
s3:
enabled: false
s3ManagedLedgerOffloadRegion: "[YOUR REGION OF S3]"
s3ManagedLedgerOffloadBucket: "[YOUR BUCKET OF S3]"
s3ManagedLedgerOffloadMaxBlockSizeInBytes: "67108864"
s3ManagedLedgerOffloadReadBufferSizeInBytes: "1048576"
s3ManagedLedgerOffloadServiceEndpoint: "http://s3.amazonaws.com"
# secret that stores AWS credentials. The secret should be created in the following
# format.
#       # kubectl -n pulsar create secret generic \       #   --from-literal=AWS_ACCESS_KEY_ID=[AWS ACCESS KEY] \       #   --from-literal=AWS_SECRET_ACCESS_KEY=[AWS SECRET KEY] \       #   [secret name]       #
# secret: [k8s secret that stores AWS credentials]
filesystem:
enabled: false
fileSystemURI: "file:///pulsar/data"
fileSystemProfilePath: "/pulsar/conf/filesystem-config.yaml"
fileDefaultFS: ""
fileTmpDir: "pulsar"
fileBuffersize: "4096"
ioSeqfileCompressBlocksize: "1000000"
ioSeqFileCompressionType: "BLOCK"
ioMapIndexInteval: "128"
packagesManagement:
storageProvider: "org.apache.pulsar.packages.management.storage.bookkeeper.BookKeeperPackagesStorageProvider"
ledgerRootPath: "/ledgers"
Pulsar: Functions Worker
templates/function-worker-configmap.yaml

functions:
component: functions-worker
useDedicatedRunner: false
functionMesh: false
functionMeshName: "function-mesh"
functionMeshServiceAccount: "default"
builtinConnectorConfigmap: ""
replicaCount: 1
functionState: false
enableCustomizerRuntime: false
runtimeCustomizerClassName: "org.apache.pulsar.functions.runtime.kubernetes.BasicKubernetesManifestCustomizer"
pulsarExtraClasspath: "extraLibs"
Specify the namespace to run pulsar functions
authenticationProviders: []
authorizationProvider: "org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider"
jobNamespace: ""
Specify the pulsar root directory
pulsarRootDir: ""
pulsarAdminUrlOverride:
pulsarServiceUrlOverride:
nodeSelector:
# cloud.google.com/gke-nodepool: default-pool

securityContext: {}
tolerations: []
extraVolumes: []
extraVolumeMounts: []
extraEnvs: []
gracePeriod: 30
Definition of the serviceAccount used to run function worker.
serviceAccount:
# Specifies whether to use a service account to run this component
use: true
# Specifies whether a service account should be created
create: true
# The name of the service account to use.
# If not set and create is true, a name is generated using the fullname template
name: ""
# Extra annotations for the serviceAccount definition. This can either be
# YAML or a YAML-formatted multi-line templated string map of the
# annotations to apply to the serviceAccount.
annotations: {}
# whether to create a cluster role
clusterRole: true
autoRollDeployment: true
annotations: {}
ports:
http: 8080
https: 8443
resources:
requests:
memory: 512Mi
cpu: 0.2
probe:
liveness:
enabled: true
failureThreshold: 10
initialDelaySeconds: 30
periodSeconds: 10
readiness:
enabled: true
failureThreshold: 10
initialDelaySeconds: 30
periodSeconds: 10
startup:
enabled: false
failureThreshold: 30
initialDelaySeconds: 60
periodSeconds: 10
service:
annotations: {}
affinity:
anti_affinity: true
# Set the anti affinity type. Valid values:
# requiredDuringSchedulingIgnoredDuringExecution - rules must be met for pod to be scheduled (hard) requires at least one node per replica
# preferredDuringSchedulingIgnoredDuringExecution - scheduler will try to enforce but not guranentee
type: preferredDuringSchedulingIgnoredDuringExecution
configData:
downloadDirectory: download/pulsar_functions
pulsarFunctionsNamespace: public/functions
functionMetadataTopicName: metadata
clusterCoordinationTopicName: coordinate
numHttpServerThreads: 8
schedulerClassName: "org.apache.pulsar.functions.worker.scheduler.RoundRobinScheduler"
functionAssignmentTopicName: "assignments"
failureCheckFreqMs: 30000
rescheduleTimeoutMs: 60000
initialBrokerReconnectMaxRetries: 60
assignmentWriteMaxRetries: 60
instanceLivenessCheckFreqMs: 30000
# Frequency how often worker performs compaction on function-topics
topicCompactionFrequencySec: 1800
# kubernetes runtime
functionRuntimeFactoryClassName: org.apache.pulsar.functions.runtime.kubernetes.KubernetesRuntimeFactory
# Connectors
connectorsDirectory: ./connectors
functionsDirectory: ./functions
narExtractionDirectory: ""
functionRuntimeFactoryConfigs:
Pulsar: pulsar detector
templates/pulsar-detector-statefulset.yaml

pulsar_detector:
component: pulsar-detector
replicaCount: 1
gracePeriod: 30
port: 9000
waitZkReady: true
annotations:
prometheus.io/scrape: "true"
prometheus.io/port: "9000"
nodeSelector:
# cloud.google.com/gke-nodepool: default-pool

tolerations: []
Definition of the serviceAccount used to run brokers.
serviceAccount:
# Specifies whether to use a service account to run this component
use: true
# Specifies whether a service account should be created
create: true
# The name of the service account to use.
# If not set and create is true, a name is generated using the fullname template
name: ""
# Extra annotations for the serviceAccount definition. This can either be
# YAML or a YAML-formatted multi-line templated string map of the
# annotations to apply to the serviceAccount.
annotations: {}
Proxy service
templates/pulsar-detector-service.yaml

service:
# spec:
# clusterIP: None
annotations: {}
Pulsar detector PodDisruptionBudget
templates/pulsar-detector-pdb.yaml

pdb:
usePolicy: true
maxUnavailable: 1
Pulsar: Proxy Cluster
templates/proxy-statefulset.yaml

proxy:
use a component name that matches your grafana configuration
so the metrics are correctly rendered in grafana dashboard
component: proxy
replicaCount: 3
nodeSelector:
# cloud.google.com/gke-nodepool: default-pool

probe:
liveness:
enabled: true
failureThreshold: 10
initialDelaySeconds: 30
periodSeconds: 10
readiness:
enabled: true
failureThreshold: 10
initialDelaySeconds: 30
periodSeconds: 10
startup:
enabled: false
failureThreshold: 30
initialDelaySeconds: 60
periodSeconds: 10
affinity:
anti_affinity: true
# Set the anti affinity type. Valid values:
# requiredDuringSchedulingIgnoredDuringExecution - rules must be met for pod to be scheduled (hard) requires at least one node per replica
# preferredDuringSchedulingIgnoredDuringExecution - scheduler will try to enforce but not guranentee
type: preferredDuringSchedulingIgnoredDuringExecution
annotations: {}
securityContext: {}
tolerations: []
gracePeriod: 30
resources:
requests:
memory: 128Mi
cpu: 0.2
extraInitContainers: []
extraVolumes: []
extraVolumeMounts: []
extraEnv: []
Definition of the serviceAccount used to run proxies.
serviceAccount:
# Specifies whether to use a service account to run this component
use: true
# Specifies whether a service account should be created
create: true
# The name of the service account to use.
# If not set and create is true, a name is generated using the fullname template
name: ""
# Extra annotations for the serviceAccount definition. This can either be
# YAML or a YAML-formatted multi-line templated string map of the
# annotations to apply to the serviceAccount.
annotations: {}
websocket:
component: websocket
enabled: false
configData:
PULSAR_MEM: >
-Xms64m -Xmx64m -XX:MaxDirectMemorySize=64m
PULSAR_GC: >
-XX:+UseG1GC
-XX:MaxGCPauseMillis=10
-Dio.netty.leakDetectionLevel=disabled
-Dio.netty.recycler.linkCapacity=1024
-XX:+ParallelRefProcEnabled
-XX:+UnlockExperimentalVMOptions
-XX:+AggressiveOpts
-XX:+DoEscapeAnalysis
-XX:ParallelGCThreads=4
-XX:ConcGCThreads=4
-XX:G1NewSizePercent=50
-XX:+DisableExplicitGC
-XX:-ResizePLAB
-XX:+ExitOnOutOfMemoryError
-XX:+PerfDisableSharedMem
Proxy configmap
templates/proxy-configmap.yaml

Automtically Roll Deployments when configmap is changed
autoRollDeployment: true
Config proxy to point to an existing broker clusters
brokerServiceURL: ""
brokerWebServiceURL: ""
brokerServiceURLTLS: ""
brokerWebServiceURLTLS: ""
configData:
PULSAR_MEM: >
-Xms64m -Xmx64m -XX:MaxDirectMemorySize=64m
PULSAR_GC: >
-XX:+UseG1GC
-XX:MaxGCPauseMillis=10
-Dio.netty.leakDetectionLevel=disabled
-Dio.netty.recycler.linkCapacity=1024
-XX:+ParallelRefProcEnabled
-XX:+UnlockExperimentalVMOptions
-XX:+AggressiveOpts
-XX:+DoEscapeAnalysis
-XX:ParallelGCThreads=4
-XX:ConcGCThreads=4
-XX:G1NewSizePercent=50
-XX:+DisableExplicitGC
-XX:-ResizePLAB
-XX:+ExitOnOutOfMemoryError
-XX:+PerfDisableSharedMem
Proxy service
templates/proxy-service.yaml

ports:
http: 8080
https: 443
# httpsServicePort:
pulsar: 6650
pulsarssl: 6651
websocket: 9090
websockettls: 9443
service:
annotations: {}
type: ClusterIP
extraSpec: {}
Proxy PodDisruptionBudget
templates/proxy-pdb.yaml

pdb:
usePolicy: true
maxUnavailable: 1
Pulsar ToolSet
templates/toolset-deployment.yaml

toolset:
component: toolset
useProxy: true
replicaCount: 1
nodeSelector:
# cloud.google.com/gke-nodepool: default-pool

annotations: {}
tolerations: []
gracePeriod: 0
resources:
requests:
memory: 256Mi
cpu: 0.1
Bastion configmap
templates/bastion-configmap.yaml

Automtically Roll Deployments when configmap is changed
autoRollDeployment: true
pulsarAdminUrlOverride:
pulsarServiceUrlOverride:
extraEnv: []
configData:
PULSAR_MEM: >
-Xms64M
-Xmx128M
-XX:MaxDirectMemorySize=128M
#############################################################
Monitoring Stack : Prometheus / Grafana
#############################################################
configmapReload:
prometheus:
## If false, the configmap-reload container will not be deployed
##
enabled: true
## configmap-reload container name
##
name: configmap-reload
## configmap-reload container image
##
image:
repository: jimmidyson/configmap-reload
tag: v0.3.0
pullPolicy: IfNotPresent
## Additional configmap-reload container arguments
##
extraArgs: {}
## Additional configmap-reload volume directories
##
extraVolumeDirs: []

## Additional configmap-reload mounts
##
extraConfigmapMounts: []
  # - name: prometheus-alerts
  #   mountPath: /etc/alerts.d
  #   subPath: ""
  #   configMap: prometheus-alerts
  #   readOnly: true

## configmap-reload resource requests and limits
## Ref: http://kubernetes.io/docs/user-guide/compute-resources/
##
resources: {}

alertmanager:
## If false, the configmap-reload container will not be deployed
##
enabled: true
## configmap-reload container name
##
name: configmap-reload

## configmap-reload container image
##
image:
  repository: jimmidyson/configmap-reload
  tag: v0.3.0
  pullPolicy: IfNotPresent

## Additional configmap-reload container arguments
##
extraArgs: {}
## Additional configmap-reload volume directories
##
extraVolumeDirs: []

## Additional configmap-reload mounts
##
extraConfigmapMounts: []
  # - name: prometheus-alerts
  #   mountPath: /etc/alerts.d
  #   subPath: ""
  #   configMap: prometheus-alerts
  #   readOnly: true

## configmap-reload resource requests and limits
## Ref: http://kubernetes.io/docs/user-guide/compute-resources/
##
resources: {}

Monitoring Stack: Prometheus
templates/prometheus-deployment.yaml

prometheus:
component: prometheus
replicaCount: 1
nodeSelector:
# cloud.google.com/gke-nodepool: default-pool

annotations: {}
tolerations: []
gracePeriod: 0
port: 9090
resources:
requests:
memory: 256Mi
cpu: 0.1
Definition of the serviceAccount used to run brokers.
serviceAccount:
# Specifies whether to use a service account to run this component
use: true
# Specifies whether a service account should be created
create: true
# The name of the service account to use.
# If not set and create is true, a name is generated using the fullname template
name: ""
# Extra annotations for the serviceAccount definition. This can either be
# YAML or a YAML-formatted multi-line templated string map of the
# annotations to apply to the serviceAccount.
annotations: {}
# specify to use a clusterRole, set to false to only allow in a single namespace
clusterRole: true
volumes:
# use a persistent volume or emptyDir
persistence: true
data:
name: data
size: 10Gi
local_storage: true
useVolumeClaimTemplates: false
# storageClassName: ""
## If the storage class is left undefined when using persistence
## the default storage class for the cluster will be used.
##
# storageClass:
# type: pd-standard
# fsType: xfs
# provisioner: kubernetes.io/gce-pd
args:
## Prometheus data retention period (default if not specified is 15 days)
##
retention: "15d"
scrapeInterval: 15s
securityContext:
runAsUser: 65534
runAsNonRoot: true
runAsGroup: 65534
fsGroup: 65534
probe:
liveness:
enabled: true
failureThreshold: 10
initialDelaySeconds: 30
periodSeconds: 10
readiness:
enabled: true
failureThreshold: 10
initialDelaySeconds: 30
periodSeconds: 10
customRelabelConfigs: []
extraScrapeConfigs: []
Prometheus service
templates/prometheus-service.yaml

service:
# expose the load balancer
# type: LoadBalancer
annotations: {}
datadog:
component: datadog
namespace: pulsar
components:
zookeeper:
enabled: false
metrics: [
""*""
]
bookkeeper:
enabled: false
metrics: [
""""
]
broker:
enabled: false
metrics: [
""_""
]
proxy:
enabled: false
metrics: [
""_*""
]
Monitoring Stack: Grafana
templates/grafana-statefulset.yaml

grafana:
component: grafana
grafana.ini:
paths:
data: /var/lib/grafana/pulsar/data
plugins: /var/lib/grafana/pulsar/plugins
provisioning: /var/lib/grafana/pulsar_provisioning
server:
domain: "{{ GRAFANA_DOMAIN }}"
serve_from_sub_path: "{{ GRAFANA_SERVE_FROM_SUB_PATH }}"
root_url: "{{ GRAFANA_ROOT_URL }}"
analytics:
check_for_updates: true
security:
admin_user: "{{ GRAFANA_ADMIN_USER }}"
admin_password: "{{ GRAFANA_ADMIN_PASSWORD }}"
auth.azuread:
name: Azure AD
enabled: false
allow_sign_up: true
client_id: ""
client_secret: ""
scopes: openid email profile
auth_url: ""
token_url: ""
allowed_domains: ""
allowed_groups: ""
role_attribute_strict: true
log:
mode: console
log.file:
level: info
format: text
grafana_com:
url: https://grafana.com
replicaCount: 1
nodeSelector:
# cloud.google.com/gke-nodepool: default-pool

annotations: {}
tolerations: []
gracePeriod: 0
port: 3000
localRenderCallback: false
resources:
requests:
memory: 250Mi
cpu: 0.1
volumes:
# use a persistent volume or emptyDir
persistence: true
## templates/grafana-statefulset.yaml
## environment variables since grafana.ini settings do not override in Docker container
## https://grafana.com/docs/grafana/latest/administration/configure-docker/#default-paths
env:
- name: GF_PATHS_DATA
value: /var/lib/grafana/pulsar/data
- name: GF_PATHS_PLUGINS
value: /var/lib/grafana/pulsar/plugin
# - name: GF_PATHS_LOGS
#   value: /var/lib/grafana/pulsar/logs
- name: GF_PATHS_PROVISIONING
value: /var/lib/grafana/pulsar_provisioning
securityContext:
# Grafana docker image user and groups: https://grafana.com/docs/grafana/latest/installation/docker/#migration-from-a-previous-version-of-the-docker-container-to-5-1-or-later
runAsUser: 472
runAsNonRoot: true
runAsGroup: 472
fsGroup: 472
mountPath: /var/lib/grafana/pulsar
data:
name: data
size: 10Gi
local_storage: true
# storageClassName: ""
## If the storage class is left undefined when using persistence
## the default storage class for the cluster will be used.
##
# storageClass:
# type: pd-standard
# fsType: xfs
# provisioner: kubernetes.io/gce-pd
Grafana service
templates/grafana-service.yaml

service:
# spec:
# type: clusterIP
annotations: {}
datasources:
loki: loki
admin:
user: pulsar
password: pulsar
Oauth2 for Azuread
Grafana Override configuration with environment variables.
see: https://grafana.com/docs/grafana/latest/administration/configuration/#override-configuration-with-environment-variables
azureAuthEnabled: false
azuread:
client_id: "AZURE-AD-SSO-CLIENT-ID"
client_secret: "AZURE-AD-SSO-CLIENT-SECRET"
Monitoring Stack: node_exporteer
templates/node-exporter.yaml

node_exporter:
component: node-exporter
annotations: {}
limits:
cpu: 10m
memory: 50Mi
requests:
cpu: 10m
memory: 50Mi
alert_manager:
component: alert-manager
port: 9093
annotations: {}
replicaCount: 1
gracePeriod: 0
resources:
requests:
memory: 250Mi
cpu: 0.1
service:
# spec:
# clusterIP: None
annotations: {}
securityContext:
runAsUser: 65534
runAsNonRoot: true
runAsGroup: 65534
fsGroup: 65534
probe:
readiness:
enabled: true
failureThreshold: 10
initialDelaySeconds: 30
periodSeconds: 10
alert manager config
config:
global:
resolve_timeout: 1m
route:
group_interval: 1m
repeat_interval: 10m
receiver: 'pagerduty-notifications'
receivers:
- name: 'pagerduty-notifications'
pagerduty_configs:
- service_key: "[PAGER DUTRY SERVICE KEY]"
send_resolved: true
add alert rules below
rules:
groups:
Components Stack: pulsar_manager
templates/pulsar-manager.yaml

pulsar_manager:
component: pulsar-manager
this can be used to indiciate to use vault auth for pulsar-manager
even if authentication isn't enabled for other components
force_vault: false
ports:
frontend: 9527
backend: 7750
replicaCount: 1
probe:
liveness:
enabled: true
failureThreshold: 10
initialDelaySeconds: 10
periodSeconds: 30
readiness:
enabled: true
failureThreshold: 10
initialDelaySeconds: 10
periodSeconds: 30
startup:
enabled: false
failureThreshold: 30
initialDelaySeconds: 10
periodSeconds: 30
nodeSelector:
# cloud.google.com/gke-nodepool: default-pool

annotations: {}
tolerations: []
gracePeriod: 0
resources:
requests:
memory: 250Mi
cpu: 0.1
volumes:
# use a persistent volume or emptyDir
persistence: true
data:
name: data
size: 10Gi
local_storage: true
useVolumeClaimTemplates: false
storageClassName: "ebs-sc"
## If the storage class is left undefined when using persistence
## the default storage class for the cluster will be used.
##
# storageClass:
# type: pd-standard
# fsType: xfs
# provisioner: kubernetes.io/gce-pd
Pulsar manager service
templates/pulsar-manager-service.yaml

service:
# expose the load balancer
# type: LoadBalancer
spec: {}
annotations: {}
ports:
frontend: 9527
backend_service:
spec:
clusterIP: None
annotations: {}
pulsar manager configmap
templates/pulsar-manager-configmap.yaml

configData: {}
superuser:
user: "pulsarmanager"
password: "welovepulsar"
description: "Pulsar Manager Admin"
email<EMAIL_ADDRESS>redirect:
host: localhost
scripts:
backend_entrypoint:
command: /pulsar-manager/pulsar-manager/bin/pulsar-manager
# extra arguments
# extraArgs:
spring:
datasource:
username: pulsar
password: pulsar
Components Stack: pulsar operators rbac
templates/pulsar-operators-rbac.yaml

rbac:
enable: true
roleName: pulsar-operator
roleBindingName: pulsar-operator-cluster-role-binding
Deploy pulsar sql
presto:
security:
authentication:
password:
enabled: false
# a K8S secret that stores the password file
#
# a) generate the password file password.db:
#    $ touch password.db
#    $ htpasswd -B -C 10 password.db
#
# b) generate the secret: kubectl -n pulsar create secret generic [PASSWORD FILE SECRET NAME] --from-file=password=password.db
#    example: kubectl -n pulsar create secret generic prod-pulsar-presto-password --from-file=password=password.db
passwordFileName: "password.db"
passwordFileSecret: '[PASSWORD FILE SECRET NAME]'
passwordFileSecretKey: 'password'
jwt:
enabled: false
publicKeyFileName: "public.key"
publicKeyConfigMapName: '[PUBLIC KEY CM NAME]'
# the public key stored in the configmap should be persisted in PEM format.
# presto prefers the PEM format. If you generate the public key using pulsar tokens CLI,
# you can use openssl rsa -pubin -in <public key file> -inform DER -pubout -out <public key file in PEM format> -outform PEM
# to convert the public key file to a public key file in PEM format.
publicKeyConfigMapKey: 'public.key'
rules: >
{
"rules": [
]
}
coordinator:
component: coordinator
replicaCount: 1
tolerations: []
affinity:
anti_affinity: true
# Set the anti affinity type. Valid values:
# requiredDuringSchedulingIgnoredDuringExecution - rules must be met for pod to be scheduled (hard) requires at least one node per replica
# preferredDuringSchedulingIgnoredDuringExecution - scheduler will try to enforce but not guranentee
type: preferredDuringSchedulingIgnoredDuringExecution
annotations: {}
gracePeriod: 10
ports:
http: 8081
https: 8443
resources:
requests:
memory: 4Gi
cpu: 2
# nodeSelector:
# cloud.google.com/gke-nodepool: default-pool
probe:
liveness:
enabled: true
failureThreshold: 10
initialDelaySeconds: 10
periodSeconds: 30
path: "/v1/status"
readiness:
enabled: true
failureThreshold: 10
initialDelaySeconds: 10
periodSeconds: 30
path: "/v1/status"
startup:
enabled: false
failureThreshold: 30
initialDelaySeconds: 10
periodSeconds: 30
config:
http:
port: 8081
query:
maxMemory: "1GB"
maxMemoryPerNode: "128MB"
maxTotalMemoryPerNode: "256MB"
jvm:
memory: 2G
jdk11: false
log:
presto:
level: DEBUG
worker:
service:
spec:
type: ClusterIP
component: worker
replicaCount: 2
tolerations: []
affinity:
anti_affinity: true
# Set the anti affinity type. Valid values:
# requiredDuringSchedulingIgnoredDuringExecution - rules must be met for pod to be scheduled (hard) requires at least one node per replica
# preferredDuringSchedulingIgnoredDuringExecution - scheduler will try to enforce but not guranentee
type: preferredDuringSchedulingIgnoredDuringExecution
annotations: {}
gracePeriod: 10
ports:
http: 8081
resources:
requests:
memory: 4Gi
cpu: 2
# nodeSelector:
# cloud.google.com/gke-nodepool: default-pool
probe:
liveness:
enabled: true
failureThreshold: 10
initialDelaySeconds: 10
periodSeconds: 30
readiness:
enabled: true
failureThreshold: 10
initialDelaySeconds: 10
periodSeconds: 30
startup:
enabled: false
failureThreshold: 30
initialDelaySeconds: 10
periodSeconds: 30
config:
query:
maxMemory: "1GB"
maxMemoryPerNode: "128MB"
maxTotalMemoryPerNode: "256MB"
jvm:
memory: 2G
jdk11: false
log:
presto:
level: DEBUG
node:
environment: production
read_offload:
enabled: false
catalog:
pulsar:
maxEntryReadBatchSize: "100"
targetNumSplits: "16"
maxSplitMessageQueueSize: "10000"
maxSplitEntryQueueSize: "1000"
namespaceDelimiterRewriteEnable: "true"
rewriteNamespaceDelimiter: "/"
bookkeeperThrottleValue: "0"
managedLedgerCacheSizeMB: "0"
bookkeeperNumIOThreads: "8"
bookkeeperNumWorkerThreads: "8"
mlNumWorkerThreads: "8"
mlNumSchedulerThreads: "8"
service:
spec:
type: ClusterIP
`

manifest
---
# Source: pulsar/templates/namespace.yaml
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#
apiVersion: v1
kind: Namespace
metadata:
  name: pulsar-poc
---
# Source: pulsar/templates/bookkeeper/bookkeeper-pdb.yaml
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#
apiVersion: policy/v1beta1
kind: PodDisruptionBudget
metadata:
  name: "pulsar-bookie"
  namespace: pulsar-poc
  labels:
    app: pulsar
    chart: pulsar-2.9.0
    release: pulsar
    heritage: Helm
    cluster: pulsar
    component: bookie
spec:
  selector:
    matchLabels:
      app: pulsar
      release: pulsar
      component: bookie
  maxUnavailable: 1
---
# Source: pulsar/templates/broker/broker-pdb.yaml
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#
apiVersion: policy/v1beta1
kind: PodDisruptionBudget
metadata:
  name: "pulsar-broker"
  namespace: pulsar-poc
  labels:
    app: pulsar
    chart: pulsar-2.9.0
    release: pulsar
    heritage: Helm
    cluster: pulsar
    component: broker
spec:
  selector:
    matchLabels:
      app: pulsar
      release: pulsar
      component: broker
  maxUnavailable: 1
---
# Source: pulsar/templates/detector/pulsar-detector-pdb.yaml
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#
apiVersion: policy/v1beta1
kind: PodDisruptionBudget
metadata:
  name: "pulsar-pulsar-detector"
  namespace: pulsar-poc
  labels:
    app: pulsar
    chart: pulsar-2.9.0
    release: pulsar
    heritage: Helm
    cluster: pulsar
    component: pulsar-detector
spec:
  selector:
    matchLabels:
      app: pulsar
      release: pulsar
      component: pulsar-detector
  maxUnavailable: 1
---
# Source: pulsar/templates/proxy/proxy-pdb.yaml
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the