Hi all... I'm debugging an issue where we're using Pulsar with non-persistent topics, and in this case, we have only 1 consumer. The data rate is low -- ~5000 msgs/sec, 1kb each -- but we see some small amount of drops. Normally, we have much larger scale and we can successfully run with 50 consumers at 3 million msgs/sec with 0 drops. So it's odd we have drops here where CPU usage on every component is quite low. The Pulsar consumer receive queue is set to 50k and never gets low. I've also taken our software out of it and replaced our consumers with the sample Java consumer code from the docs. There, we just throw away the message (no

println()

or other work per message). Again, CPU usage is under 10% for all components, but I see the same small drops. I started debugging and found Pulsar is dropping because the Netty connection's

.isWritable()

returns false and this causes Pulsar to immediately drop. This "Returns

true

if and only if the I/O thread will perform the requested write operation immediately", meaning there is room available in Netty's ChannelOutboundBuffer. I found that if I increase the Netty low/highWaterMarks, the drops go away, but it's not possible without a code change to Pulsar broker. I'm looking for any suggestions on different configurations I can try. Thanks!!

Can you provide more details about the Pulsar version, JVM settings, etc. that you are using? What are your broker settings?

Hi @David K, I see the same issue in Pulsar 2.9.1 and also Pulsar 4.0.6. We are using the Pulsar Helm chart and the settings in the configmap are:

PULSAR_GC: >
    -XX:+UseG1GC -XX:MaxGCPauseMillis=10 -Dio.netty.leakDetectionLevel=disabled
    -Dio.netty.recycler.linkCapacity=1024 -XX:+ParallelRefProcEnabled
    -XX:+UnlockExperimentalVMOptions -XX:+DoEscapeAnalysis
    -XX:ParallelGCThreads=4 -XX:ConcGCThreads=4 -XX:G1NewSizePercent=50
    -XX:+DisableExplicitGC -XX:-ResizePLAB -XX:+ExitOnOutOfMemoryError
    -XX:+PerfDisableSharedMem
  PULSAR_MEM: |
    -Xms128m -Xmx4G -XX:MaxDirectMemorySize=4G
  brokerDeleteInactiveTopicsEnabled: 'false'
  brokerServicePort: '6650'
  clusterName: aa-pulsar
  configurationStoreServers: aa-pulsar-zookeeper:2181
  defaultRetentionSizeInMB: '-1'
  defaultRetentionTimeInMinutes: '60'
  dispatcherMaxReadBatchSize: '1000'
  exposeConsumerLevelMetricsInPrometheus: 'true'
  exposeTopicLevelMetricsInPrometheus: 'true'
  functionsWorkerEnabled: 'false'
  managedLedgerDefaultAckQuorum: '1'
  managedLedgerDefaultEnsembleSize: '1'
  managedLedgerDefaultWriteQuorum: '1'
  maxConcurrentNonPersistentMessagePerConnection: '10000'
  numHttpServerThreads: '8'
  statusFilePath: /pulsar/logs/status
  webServicePort: '8080'
  zooKeeperSessionTimeoutMillis: '30000'
  zookeeperServers: aa-pulsar-zookeeper:2181

Settings for topics, etc. are the defaults. In my test, I am producing data to 7 non-persistent topics. I see the same behavior if I subscribe to all 7 (multi-topic subscription) or just 1.

“I started debugging and found Pulsar is dropping because the Netty connection’s

.isWritable()

returns false.” Can you share the code where you found this?

Hi @David K, sure! The entry point is here in NonPersistentDispatcherMultipleConsumers::sendMessages(): https://github.com/apache/pulsar/blob/88287345d3246fe0d6ea34e06389356ada516cfa/pul[…]ice/nonpersistent/NonPersistentDispatcherMultipleConsumers.java I should have mentioned all our topics are

shared

key type, so that means the *MultipleConsumers dispatcher. From here, we go to AbstractDispatcherMultipleConsumers::getNextConsumer(): https://github.com/apache/pulsar/blob/88287345d3246fe0d6ea34e06389356ada516cfa/pul[…]/pulsar/broker/service/AbstractDispatcherMultipleConsumers.java And then to ADMC::getNextConsumerFromSameOrLowerLevel() which returns null because

isConsumerAvailable()

returns false. That eventually calls through to the Netty

.isWritable()

method I mentioned which is returning false. https://github.com/apache/pulsar/blob/88287345d3246fe0d6ea34e06389356ada516cfa/pul[…]/pulsar/broker/service/AbstractDispatcherMultipleConsumers.java