Question around performance of the Producer

[wmorgan6796]

Hi,

My team and I have been using librdkafka in our C++ service quite happily and we've recently noticed something. We're doing stream processing at quite a large scale (anywhere from ~20 million messages a second) and we've noticed that no matter what we do we reach a point where the producer part of this library can't drain its outgoing queue fast enough. We've set the batch sizes and queue sizes quite large, but it just doesn't seem to be able to keep up and we end up receiving a ton of Queue full errors. The way we've gotten around this for now is creating multiple producers within a single library (I think we use 8 today since that seems to be able to keep up with a bit of leeway). This of course comes with the tradeoff of spawning many threads since we're now doing Num of brokers x Num of producers which is starting to get unwieldy (as you can imagine, the size of the Kafka cluster to keep up with that many messages a second at scale is quite large) My question is this:

What can we tune so that the outgoing queue on the producer can be drained faster? What we have now works, but there are limits to the number of threads a host can support and we would prefer to not hit that.

Thanks!

[edenhill]

Hi,

some detailing questions:

• what librdkafka version are you on?
• what's your produder configuration?
• what's the CPU usage of librdkafka threads when it hits the wall?
• how many topics are your producing to, and how many partitions do each of those topics have? How many of those partitions are typically handled by a single broker?
• typical message sizes? are you using compression, and if so which compression type, and do the messages compress well?
• are you monitoring librdkafka statistics?

[wmorgan6796]

I'll provide what I can:
Version: 1.9.2

Producer Configuration:

===== Global Kafka Producer Config =====

        allow.auto.create.topics = false
        api.version.fallback.ms = 0
        api.version.request = true
        api.version.request.timeout.ms = 10000
        auto.commit.interval.ms = 5000
        batch.num.messages = 1000000
        batch.size = 1000000
        broker.address.family = any
        broker.address.ttl = 1000
        broker.version.fallback = 0.10.0
        builtin.features = gzip,snappy,ssl,sasl,regex,lz4,sasl_gssapi,sasl_plain,sasl_scram,plugins,zstd,sasl_oauthbearer,http,oidc
        check.crcs = false
        client.id = resequencer.producer
        client.rack =
        client.software.name = librdkafka
        compression.codec = lz4
        connections.max.idle.ms = 0
        coordinator.query.interval.ms = 600000
        debug =
        default_topic_conf = 0x7f7df0981080
        delivery.report.only.error = false
        enable.auto.commit = true
        enable.auto.offset.store = true
        enable.gapless.guarantee = false
        enable.idempotence = false
        enable.partition.eof = false
        enable.random.seed = true
        enable.sasl.oauthbearer.unsecure.jwt = false
        enable.sparse.connections = true
        enable.ssl.certificate.verification = true
        enable_sasl_queue = false
        enabled_events = 0
        fetch.error.backoff.ms = 500
        fetch.max.bytes = 52428800
        fetch.message.max.bytes = 1048576
        fetch.min.bytes = 1
        fetch.wait.max.ms = 500
        group.protocol.type = consumer
        heartbeat.interval.ms = 3000
        internal.termination.signal = 0
        isolation.level = read_committed
        log.connection.close = true
        log.queue = false
        log.thread.name = true
        log_cb = 0x6ab040
        log_level = 6
        max.in.flight.requests.per.connection = 1000000
        max.poll.interval.ms = 300000
        message.copy.max.bytes = 65535
        message.max.bytes = 1000000
        message.send.max.retries = 2147483647
        metadata.broker.list = <redacted>
        metadata.max.age.ms = 900000
        metadata.request.timeout.ms = 10
        offset.store.method = broker
        open_cb = 0x720600
        partition.assignment.strategy = range,roundrobin
        queue.buffering.backpressure.threshold = 1
        queue.buffering.max.kbytes = 8000000
        queue.buffering.max.messages = 1000000
        queue.buffering.max.ms = 5000
        queued.max.messages.kbytes = 65536
        queued.min.messages = 100000
        receive.message.max.bytes = 100000000
        reconnect.backoff.jitter.ms = 0
        reconnect.backoff.max.ms = 10000
        reconnect.backoff.ms = 100
        retry.backoff.ms = 100
        sasl.kerberos.kinit.cmd = kinit -R -t "%{sasl.kerberos.keytab}" -k %{sasl.kerberos.principal} || kinit -t "%{sasl.kerberos.keytab}" -k %{sasl.kerberos.principal}
        sasl.kerberos.min.time.before.relogin = 60000
        sasl.kerberos.principal = kafkaclient
        sasl.kerberos.service.name = kafka
        sasl.mechanisms = GSSAPI
        sasl.oauthbearer.method = default
        security.protocol = plaintext
        session.timeout.ms = 45000
        socket.blocking.max.ms = 1000
        socket.connection.setup.timeout.ms = 30000
        socket.keepalive.enable = false
        socket.max.fails = 1
        socket.nagle.disable = false
        socket.receive.buffer.bytes = 0
        socket.send.buffer.bytes = 0
        socket.timeout.ms = 60000
        socket_cb = 0x6c6540
        ssl.ca.certificate.stores = Root
        ssl.endpoint.identification.algorithm = none
        ssl.engine.id = dynamic
        statistics.interval.ms = 60000
        sticky.partitioning.linger.ms = 10
        test.mock.broker.rtt = 0
        test.mock.num.brokers = 0
        topic.metadata.propagation.max.ms = 30000
        topic.metadata.refresh.fast.cnt = 10
        topic.metadata.refresh.fast.interval.ms = 250
        topic.metadata.refresh.interval.ms = 300000
        topic.metadata.refresh.sparse = true
        transaction.timeout.ms = 60000
===== Topic Kafka Producer Config =====

        auto.commit.enable = false
        auto.commit.interval.ms = 60000
        auto.offset.reset = largest
        compression.codec = inherit
        compression.level = -1
        consume.callback.max.messages = 0
        message.timeout.ms = 300000
        offset.store.method = broker
        offset.store.path = .
        offset.store.sync.interval.ms = -1
        partitioner = consistent_random
        produce.offset.report = false
        queuing.strategy = fifo
        request.required.acks = 1
        request.timeout.ms = 30000

Cpu Usage is 40% on a 32 Core, 128GB machine

Producing to about ~2000 Topics with the data evenly going to all of them, each topic is a single partition.

Typical message size is less than 100KB uncompressed and we use LZ4. Messages compress reasonably well (saw about 50% drop in message size when compression is enabled)

We are monitoring librdkafka statistics but not everything.

Thanks

[wmorgan6796]

Also important to note is that we use plaintext protocol

[wmorgan6796]

My gut says we've set our linger.ms too high for the expected throughput. We might be one of the few cases where it being set to a higher number doesn't make as much sense since we're able to actually fill our queue in that time

[wmorgan6796]

As an update on this, it seems that moving linger.ms down to 1 second helps a bit, but not when also using less producers.

[edenhill]

We typically see performance improvements to drop off at a linger.ms above 500ms.

As for the overall performance and CPU cost, there's a problem with the way librdkafka sends ProduceRequests to brokers in that it will only send batches for a single partition per request, while the protocol and Java producer will send batches for multiple partitions in the same request.
This is typically not a problem, unless you are producing to a higher number of partitions since batches will only accumulate per partition thus leading to ProduceRequests that are nowhere near the maximum request size.

[wmorgan6796]

Is there a way to see if this is the cause? I’m assuming there’s probably a statistic somewhere. We are pushing to a large number of partitions across about 400 brokers.

In short it seems like currently there is no fix for the performance issues due to inherent design decisions made in librdkafka?

[wmorgan6796]

I would be willing to take a crack at making the change to be like the Java library, but I’m not sure if there was a reason for that design choice nor where I would start.