TGIP-CN-009：Pulsar performance tuning

1 .TGIP-CN 009 Apache Pulsar performance tuning Penghui Li 

2 .Pulsar components Client Broker Bookie Pulsar SQL Function IO Connector Pulsar Proxy Metadata Store Tiered Storage 

3 .Outline 1.Basic concepts 2.Message writing 3.Message reading 

4 .Basic Concepts 

5 .Basic concepts Topic - Producer - Consumer 5 4 3 2 1 0 Producer Topic Consumer 

6 .Basic concepts Topic - Managed Ledger Topic Managed Ledger Ledger Ledger Ledger ō Ledger Ledger Ledger Cursor Cursor Cursor ō Cursor Cursor Cursor 

7 .Basic concepts Managed Ledger Cache Read from Cache Read from Bookies Topic Tailing Cache 

8 .Basic concepts Ledger - Entry Topic Ledger - 2 (open) Ledger - 1 (closed) Ledger - 0 (closed) Ledger rollover Fragment F - 1 F - 0 Fragment Bookie failover E-3 E-2 E-1 E-0 E-3 E-2 E-1 E-0 E-3 E-2 E-1 E-0 

9 .Basic concepts Journal - Entry log L-0 E-0 L-1 E-0 L-2 E-0 L-0 E-0 L-0 E-0 L-1 E-0 L-1 E-1 Checkpoint L-1 E-0 L-1 E-1 L-1 E-2 L-1 E-1 L-2 E-0 L-0 E-1 L-2 E-0 Journal Entry Log - 0 Index DB Entry Log - 1 

10 .Basic concepts Message Single Message Batched Message Client 2 Client 6 7 8 Broker 1 Broker 3 4 5 Bookie 0 Bookie 0 1 2 

11 .Basic concepts Message - Entry Entry-3 Entry-2 Entry-1 Entry-0 8 7 6 5 4 3 2 1 0 11 10 9 

12 .Message writing 

13 .Topic Producer 0 1 2 3 4 Message route mode Partition switch P-0 P-1 P-2 P-3 P-4 Partitioned Topic 

14 .Producer pending queue Producer Broker . 7 6 5 4 3 2 1 0 4 3 2 1 0 Send response to the client . 12 11 10 9 8 7 6 5 

15 .Producer pending queue Producer Broker 8 7 6 5 4 3 2 1 0 2 1 0 Send response to the client ….. 8 7 6 5 4 3 

16 .Publish batch message Producer<byte[]> producer = client.newProducer() .topic("my-topic") .enableBatching(true) .batchingMaxBytes(128 * 1024 * 1024) .batchingMaxMessages(1000) .batchingMaxPublishDelay(2, TimeUnit.MILLISECONDS) .blockIfQueueFull(true) .roundRobinRouterBatchingPartitionSwitchFrequency(10) .batcherBuilder(BatcherBuilder.DEFAULT) .create(); producer.newMessage().value(new byte[0]).send(); producer.newMessage().value(new byte[0]).sendAsync(); producer.send(new byte[0]); producer.sendAsync(new byte[0]); 

17 .Message compression Producer<byte[]> producer = client.newProducer() .topic("my-topic") .compressionType(CompressionType.LZ4) .create(); LZ4 ZLIB ZSTD SNAPPY All versions >= 2.3.0 >= 2.4.0 

18 .Message writing overview Topic ō Brokers ō Bookies ō 

19 .E Qw Qa Broker Broker 0 0 0 0 0 0 1 1 1 1 1 1 2 2 2 2 2 2 3 3 3 3 3 3 4 4 4 4 4 4 Bookie-0 Bookie-1 Bookie-2 Bookie-0 Bookie-1 Bookie-2 Bookie-3 Bookie-4 3-3-2 5-3-2 

20 .Bookie handle entry add Request processor 1.Append into Journal 3.Complete add request Journal Memory Table 2.Add into Ledger Storage 4.Periodically flush Entry Log Ledger Storage 

21 .Broker configurations 1. managedLedgerDefaultEnsembleSize 2. managedLedgerDefaultWriteQuorum 3. managedLedgerDefaultAckQuorum 4. managedLedgerNumWorkerThreads 5. numIOThreads 6. -Dorg.apache.bookkeeper.conf.readsystemproperties=true -DnumIOThreads=8 

22 .Bookie configurations 1. Journal Directories 2. Ledger Directories 3. Journal sync data 4. Journal group commit 5. Write cache 6. Flush interval 7. Add worker threads and max pending add requests 8. Journal pagecache flush interval 

23 .Message reading 

24 .Consumer receiver queue 9 8 7 6 5 4 3 2 1 0 1 2 9 8 7 6 5 14 13 12 11 10 9 8 3 4 

25 .Message reading overview Topic ō Brokers ō Bookies ō 

26 .Message dispatcher Consumer Managed ledger Dispatcher Consumer Consumer 

27 .Key_Shared dispatcher 65535 Consumer - 0 Consumer - 3 1.Peek key 2.Get key hash 3.Group by hash % slots 3 2 0 29093 Consumer - 2 Consumer - 5 6 5 4 3 2 1 0 6 4 10297 Consumer - 4 5 1 507 Consumer - 6 Consumer - 1 0 

28 .Key based batcher <k0-v1> <k0-v0> <k0-v1> <k1-v1> <k1-v0> <k0-v0> <k1-v1> <k1-v0> partition_key ordering_key 

29 .Bookie configurations 1. dbStorage_rocksDB_blockCacheSize 2. dbStorage_readAheadCacheMaxSizeMb 3. dbStorage_readAheadCacheBatchSize 4. Read worker threads