Librdkafka: 0.11 consumer memory usage and slow performance

[greatt1n]

Hi,I used high-level consumer api and used rd_kafka_consume_batch_queue() method.
I found that the memory goes up quickly when I just only started the consumer ,not consume any messages . The memory can go up to 21g . It seems that there is one pthread background to fetch messages continually and put them in buffer until EOF. Topic I fetched has 4 million messages and one message has about 1K size.
So how can i limit the memory usage?

[edenhill]

What is your client configuration?

[greatt1n]

Most client configuration is default.
I seted "queued.max.messages.kbytes=10000" and the memory brought down obviously.
It is strange that memory usage is normal with kafka 0.10 but goes up crazily after upgrading kafka 0.11.

[edenhill]

Which broker version is okay/bad?
Which librdkafka version is okay/bad?
Which combination of the two are okay and which is bad?

[greatt1n]

kafka_2.11-0.10.0.1 + rdkafka 0.9.5 is ok.
kafka_2.11-0.10.0.1 + rdkafka 0.11.0 is ok.
kafka_2.11-0.11.0.0 + rdkafka 0.9.5 is bad.
kafka_2.11-0.11.0.0 + rdkafka 0.11.0 is bad.

[edenhill]

Are you destroying all the messages you get back from consume..() in a timely manner?
See https://github.com/edenhill/librdkafka/wiki/FAQ#explain-the-consumers-memory-usage-to-me

[greatt1n]

Yes,I destroyed all the messages with the function rd_kafka_message_destroy().

[edenhill]

What OS is this?
Are you looking at VIRT or RSS of the process?
Any chance you could run the program with valgrind to find out where the memory is allocated?

[greatt1n]

Okay,let me try the valgrind.
Linux OS is Red Hat 7.2 and I read the RES with the "top" tool.

[tchap]

Hi,

I am having a similar issue, I guess.

I use librdkafka with Go and with queued.max.messages.kbytes=5000 my process is exhausting all 512 MB RAM I assigned to it. By using standard Go tools I discovered that Go is eating between 30 and 50 MB RAM, so it is definitely not my code leaking.

What am I doing wrong? Is there some other option I need to set?

BTW using https://github.com/edenhill/librdkafka/commit/28bacbabc5b29b4179553fd901df7d391997238d

I will probably try valgrind tomorrow.

[tchap]

Also it should be noted that the consumer is pretty slow since it needs to initially fill an internal cache that takes some time, so it is processing like 3 messages per second at the beginning.

[tchap]

And yeah, the consumer is fetching like 20 MBps of data in the background. This all totally leads me to think that there is some buffer not limited in the right way. Not sure why queued.max.messages.kbytes is not working...

[tchap]

Hmm, Valgrind is having issues with Go programs. Not sure I can run it at all.

[edenhill]

What's your client configuration?

[tchap]

@edenhill

{  
   "bootstrap.servers":"...",
   "client.id":"...",
   "default.topic.config":{  
      "auto.offset.reset":"largest"
   },
   "enable.auto.commit":false,
   "go.events.channel.enable":true,
   "go.events.channel.size":1000,
   "group.id":"...",
   "queued.max.messages.kbytes":10000,
   "queued.min.messages":1000
}

I see that I actually set the option to 10000, but that is still 10 MB per partion per topic, which should add to 120 MB in my case...

It takes about 30 seconds for the consumer to start consuming the memory. The first 30 seconds it's not doing anything suspicious really.

[edenhill]

What's your typical message size?

[tchap]

About 200 B. Less than a KB definitely.

[edenhill]

Okay, is there a fixed number of messages in the topic or is it being constantly produced to? If so, at what rate?

[tchap]

Constantly produced, 300 msg/s that are partitioned into 12 partitions. This consumer is handling all 12 partitions at the moment.

[tchap]

But this consumer is probably lagging behind the producer terribly since it is crashing at the moment. I only start it for testing purposes occasionally.

[tchap]

I mean, it is probably starting with the first message that is available in the topic. We retain messages for 24 hours, so it is probably starting with the oldest message available.

[edenhill]

Is this reproducible with examples/consumer_channel_example ?

[tchap]

Trying now using the following code:

package main

import (
    "fmt"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/confluentinc/confluent-kafka-go/kafka"
)

func main() {
    sigchan := make(chan os.Signal, 1)
    signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)

    c, err := kafka.NewConsumer(&kafka.ConfigMap{
        "bootstrap.servers":               "...",
        "client.id":                       "...",
        "group.id":                        "...",
        "session.timeout.ms":              6000,
        "enable.auto.commit":              false,
        "go.events.channel.enable":        true,
        "go.events.channel.size":          10,
        "go.application.rebalance.enable": true,
        "default.topic.config":            kafka.ConfigMap{"auto.offset.reset": "largest"},
        "queued.max.messages.kbytes":      10000,
        "queued.min.messages":             1000,
    })

    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to create consumer: %s\n", err)
        os.Exit(1)
    }

    fmt.Printf("Created Consumer %v\n", c)

    err = c.Subscribe("data.messages", nil)

    run := true

    commitTicker := time.NewTicker(5 * time.Second)
    defer commitTicker.Stop()

    partitions := make(map[int32]kafka.TopicPartition, 12)

    for run == true {
        select {
        case sig := <-sigchan:
            fmt.Printf("Caught signal %v: terminating\n", sig)
            run = false

        case ev := <-c.Events():
            switch e := ev.(type) {
            case kafka.AssignedPartitions:
                fmt.Fprintf(os.Stderr, "%% %v\n", e)
                c.Assign(e.Partitions)
            case kafka.RevokedPartitions:
                fmt.Fprintf(os.Stderr, "%% %v\n", e)
                c.Unassign()
            case *kafka.Message:
                fmt.Printf("%% Message on %s:\n%s\n",
                    e.TopicPartition, string(e.Value))

                time.Sleep(1 * time.Second)

                partitions[e.TopicPartition.Partition] = e.TopicPartition

            case kafka.PartitionEOF:
                fmt.Printf("%% Reached %v\n", e)
            case kafka.Error:
                fmt.Fprintf(os.Stderr, "%% Error: %v\n", e)
                run = false
            }

        case <-commitTicker.C:
            offsets := make([]kafka.TopicPartition, 0, len(partitions))
            for _, partition := range partitions {
                offsets = append(offsets, partition)
            }

            _, err := c.CommitOffsets(offsets)
            if err != nil {
                fmt.Fprintf(os.Stderr, "%% Commit error: %v\n", err)
            } else {
                fmt.Println("%% Offsets committed %%")
            }

            partitions = make(map[int32]kafka.TopicPartition, 12)
        }
    }

    fmt.Printf("Closing consumer\n")
    c.Close()
}

The funny part is that this works just fine, but as soon as I comment out "go.application.rebalance.enable": true, the memory is consumed immediately.

[edenhill]

Does it still happen if you remove the Sleep(1) from the Message handler?
And is that making a difference with rebalance.enable?
What if you remove the CommitOffsets?

[tchap]

rebalance.enabled false && no sleep => ok
rebalance.enabled true && no sleep => ok
rebalance.enabled false && sleep && no CommitOffsets => memory consumed immediately
rebalance.enabled true && sleep && no CommitOffsets => memory consumed immediately

[tchap]

rebalance.enabled true && auto.commit true && sleep => memory consumed initially, but then it is freed and does not go up that much again
rebalance.enabled false && auto.commit true && sleep => memory consumed immediately, but then it is freed, then again consumed, then again freed, so the process keeps running; perhaps the same as the one before, cannot tell the exact difference

[tchap]

Actually
rebalance.enabled false && sleep && no CommitOffsets => also consumes and then frees
rebalance.enabled true && sleep && no CommitOffsets => also consumes and then frees

[tchap]

ACTUALLY it seems that just commenting out rebalance.enabled also just causes the memory to be consumed initially, but then it is freed after a few seconds... Hmm. Not sure what to take from all this...

[edenhill]

Do you think this could be a GC issue?
Maybe try some of the calls here:
https://golang.org/pkg/runtime/debug/

[tchap]

Just to sum up, we can be playing with various combinations, but what struck me at the beginning was that the memory actually went up over any limits. Providing the options I provided should never allow librdkafka to eat all memory, right?

[edenhill]

Do we have any idea if this is Go or librdkafka memory?

[tchap]

Ok, I can try with the example above. But for the production component Go was eating like 50 MB while the whole process was eating 512 MB.

[tchap]

That is why I actually went to post the issue here and not into the Go repo :-)

[tchap]

Yeah, added runtime.ReadMemStats during every offset commit and got

{  
   "sys_mbytes":2,
   "heap_alloc_mbytes":0,
   "heap_sys_mbytes":1,
   "heap_idle_mbytes":0,
   "heap_inuse_mbytes":1,
   "heap_released_mbytes":0,
   "stack_inuse_mbytes":0,
   "stack_sys_mbytes":0
}

even though the process was eating all memory allocated to it.

[edenhill]

Okay, cool, so then we know Go isn't the problem.

Can you provide the most minimal reproducing consumer code and I'll try to reproduce it inhouse again?

[tchap]

package main

import (
    "fmt"
    "os"
    "os/signal"
    "runtime"
    "syscall"
    "time"

    "go.uber.org/zap"
    "github.com/confluentinc/confluent-kafka-go/kafka"
)

func main() {
    sigchan := make(chan os.Signal, 1)
    signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)

    c, err := kafka.NewConsumer(&kafka.ConfigMap{
        "bootstrap.servers":        "...",
        "client.id":                "...",
        "group.id":                 "...",
        "session.timeout.ms":       6000,
        "enable.auto.commit":       false,
        "go.events.channel.enable": true,
        "go.events.channel.size":   10,
        //"go.application.rebalance.enable": true,
        "default.topic.config":       kafka.ConfigMap{"auto.offset.reset": "largest"},
        "queued.max.messages.kbytes": 10000,
        "queued.min.messages":        1000,
    })

    if err != nil {
        fmt.Fprintf(os.Stderr, "Failed to create consumer: %s\n", err)
        os.Exit(1)
    }

    fmt.Printf("Created Consumer %v\n", c)

    err = c.Subscribe("data.messages", nil)

    run := true

    commitTicker := time.NewTicker(5 * time.Second)
    defer commitTicker.Stop()

    partitions := make(map[int32]kafka.TopicPartition, 12)

    logger, _ := zap.NewProduction()

    for run == true {
        select {
        case sig := <-sigchan:
            fmt.Printf("Caught signal %v: terminating\n", sig)
            run = false

        case ev := <-c.Events():
            switch e := ev.(type) {
            case kafka.AssignedPartitions:
                fmt.Fprintf(os.Stderr, "%% %v\n", e)
                c.Assign(e.Partitions)
            case kafka.RevokedPartitions:
                fmt.Fprintf(os.Stderr, "%% %v\n", e)
                c.Unassign()
            case *kafka.Message:
                fmt.Printf("%% Message on %s:\n%s\n",
                    e.TopicPartition, string(e.Value))

                time.Sleep(1 * time.Second)

                partitions[e.TopicPartition.Partition] = e.TopicPartition

            case kafka.PartitionEOF:
                fmt.Printf("%% Reached %v\n", e)
            case kafka.Error:
                fmt.Fprintf(os.Stderr, "%% Error: %v\n", e)
                run = false
            }

        case <-commitTicker.C:
            offsets := make([]kafka.TopicPartition, 0, len(partitions))
            for _, partition := range partitions {
                offsets = append(offsets, partition)
            }

            _, err := c.CommitOffsets(offsets)
            if err != nil {
                fmt.Fprintf(os.Stderr, "%% Commit error: %v\n", err)
            } else {
                fmt.Println("%% Offsets committed %%")
            }

            partitions = make(map[int32]kafka.TopicPartition, 12)

            var stats runtime.MemStats
            runtime.ReadMemStats(&stats)

            mb := func(bytes uint64) uint64 {
                return bytes / 1024 / 1024
            }

            // Print them into the console.
            logger.Info(
                "memory stats gathered",
                zap.Uint64("sys_mbytes", mb(stats.Sys)),
                zap.Uint64("heap_alloc_mbytes", mb(stats.HeapAlloc)),
                zap.Uint64("heap_sys_mbytes", mb(stats.HeapSys)),
                zap.Uint64("heap_idle_mbytes", mb(stats.HeapIdle)),
                zap.Uint64("heap_inuse_mbytes", mb(stats.HeapInuse)),
                zap.Uint64("heap_released_mbytes", mb(stats.HeapReleased)),
                zap.Uint64("stack_inuse_mbytes", mb(stats.StackInuse)),
                zap.Uint64("stack_sys_mbytes", mb(stats.StackSys)),
            )

        }
    }

    fmt.Printf("Closing consumer\n")
    c.Close()
}

I left all the stuff in so that you can experiment with not using manual commit and stuff as I did above. But the code as passed in should consume all memory at least initially.

[edenhill]

Thank you, reproduced

[edenhill]

I've found the issue:

As described here the entire FetchResponse buffer memory will remain until all messages that reference it are destroyed.

The new FetchResponse parsing code in librdkafka for Apache Kafka 0.11.0 has a bug where it just reads the first returned MessageSet (batch) per partition, but the broker may send multiple MessageSets. In this particular case where the producer constructed single-message MessageSets, about a thousand MessageSets were returned per partition in the FetchResponse but due to the afforementioned bug only the first MessageSet's single message was used, the rest discarded.
The internal fetch queue would thus increase by one for each partition, rather than the actual 1000 messages that were available in the response.
librdkafka attempts to keep at least queued.min.messages in the queue, so it sends another FetchRequest with offset advanced by +1, this one also returning 1000 single-message MessageSets where only the first one is actually used.
Rinse repeat.

With this behaviour and the late freeing of FetchResponse memory, we end up with a situation where each enqueued message actually consumes about a MB memory (all 1000 MessageSets returned in that FetchResponse). Multiply that by the number of enqueued messages (>1000) and we quickly get a gigs worth of memory.

I'm working on a fix to read all MessageSets in the response, which will fix this memory issue and dramatically speed up the consumer for tiny MessageSets, reported in:

• #1401
• #1395 (possibly)
• #1385
• #1344
• https://github.com/confluentinc/confluent-kafka-dotnet/issues/285 (possibly)

Huge thanks in helping find the root cause for this issue! :heart:

[edenhill]

Please build latest librdkafka master and give it a spin and see if the fix is working for you.
Thanks

[tchap]

WFM ✌️

[yacut]

@edenhill I've tested latest librdkafka master with alpine linux docker image.
v. 0.11.0-RC1-82-g5aa16e, e. gzip, snappy, ssl, sasl, regex, lz4, sasl_gssapi, sasl_plain, sasl_scram, plugins.

One consumer with default settings uses 1,2-2GB RAM.
One consumer with queued.min.messages=1000 and queued.max.messages.kbytes=10000 settings uses 250-400 MB RAM. The most interesting part here is that process initial goes to 400-450MB and after 5-10 minutes stands to 280MB.

[edenhill]

@yacut I'm not sure what version 0.11.0-RC1-82-g5aa16e is as I can't find that commit in my tree, do you have any idea?

[yacut]

@edenhill I've build it a few hours ago:

bash-4.3# pwd
/root/librdkafka
bash-4.3# git rev-parse HEAD
5aa16e8852495a5685e5c4396d0defcddb21c38f

Do you need my dockerfile or another info?

[edenhill]

Whoops, my bad, there it is.

Make sure to read this:
https://github.com/edenhill/librdkafka/wiki/FAQ#explain-the-consumers-memory-usage-to-me

[yacut]

yes, I have read it, therefore also second test.
I'm just wondering that process needs double memory at start but not after 5-10 minutes, although message rate remains the same (10 million messages in the topic and ~100m/s rate).

[edenhill]

@yacut is this reproducible with rdkafka_performance -b .. -G mygrp -t mytopic .. or rdkafka_consumer_example .. ?

[tudor]

I think we're hitting this too. We have a new use case where the (Java) producers write much larger batches (~1000 messages instead of ~1) and the (C++) consumers are using up about 500x more network bandwidth than they should, given the speed at which the messages are processed downstream. Looking forward to trying this out.

[tudor]

And yes, the fix worked for us too, like a charm. Thanks!

[japdt]

Hi, is there a plan to release a version with this fix? Thanks!

[edenhill]

A release candidate for librdkafka is now available on NuGet, please help verifying the fix:
https://www.nuget.org/packages/librdkafka.redist/0.11.1-RC1

[zapacila89]

Yup. librdkafka.redist/0.11.1-RC1 fixes the memory leak. At least for Confluent Kafka DotNet.

[vin-d]

Hi, I came here after reading #1385
I've begun some tests using latest release https://github.com/edenhill/librdkafka/releases/tag/v0.11.3 and i'm currently using v0.9.5.

I'm seeing performance issue with the new release with an high level consumer :
i'm basically reading messages (average 2.7 kB uncompressed - but they are snappied on kafka) and I see that I'm limited at reading approx 300msg/sec per assigned partition

so if I have 2 partitions assigned I'm able to consume about 600 msg/sec
, if I have 3 partitions assigned I'm able to consume about 900 msg/sec, etc...

while trying with librdkafka 0.9.5 I don't have this performance problem related to the number of partitions assigned.
I can read more than 900msg/sec from one partition assigned.

does that ring a bell to anyone ?
(Kafka server 10.1.0)

[edenhill]

Is that message rate including your processing, or just consumption?

[vin-d]

just consumption.
I have my consume loop in which I reference the message in a queue to be handled by a separate thread.

[edenhill]

Even 900 msgs/sec sounds very slow (that's about 2MB/s), why is that?

[vin-d]

even less in fact i've used nethogs to see that bandwidth fluctuates around 300kB for 300msg/s rate
(so I assume snappied msg must be about 1kB each ).

But that is another issue : my highg level consumer is cosuming from one site to another site through VPN with limited bandwith so that's doesn't really matters Is my test.

what I just see from a high level monitoring is that if I assign N partitions I have a throughput of 300N messages.

while this is not true with v0.9.5 where I can go around 1500 - 2000 msg/sec with one partition assigned.

[edenhill]

That is weird indeed.
Enable debug=fetch,protocol and provide a log excerpt (some hundred lines) of a slow consumer

[vin-d]

is that enough ?
https://drive.google.com/file/d/1sXGhajiBmbZzOLYYhRs212NIxOq46OJ2/view?usp=sharing

[vin-d]

And here is the same with lirdkafka v0.9.5 https://drive.google.com/open?id=1D8hgLRN8O82w7gc_BLnoT272OkSeAdyz
where we don't see any

Protocol parse failure at 990220/1048617 (rd_kafka_msgset_reader_msg_v0_1:464) (incorrect broker.version.fallback?)

nor

2|7|BACKOFF|0|Success|[thrd:server2.private:9092/bootstrap]: server2.private:9092/1: TOPICNAME [0]: Fetch backoff for 500ms: Local: Bad message format

so I assume it's a client misconfiguration issue somewhere at my level :-/
but isn't that funny that it can still transfer message even if misconfigured with the server.

[edenhill]

Thanks, will investigate.
As a workaround you can set fetch.error.backoff.ms to a lower value than 500ms.

[vin-d]

in the meantime (I don't know if you'll consider that as a bug or as a misconfiguration from me),
I've retried a quick test with v0.11.3 and the configuration api.version.request=false on my high level consumer.
Then I see the same performance as with v0.9.5
and I can read from the lib debug logs :

2|7|APIVERSION|0|Success|[thrd:server2.private:9092/bootstrap]: server2.private/bootstrap: Using (configuration fallback) 0.9.0 protocol features
2|7|APIVERSION|0|Success|[thrd:server3.private:9092/bootstrap]: server3.private/bootstrap: Using (configuration fallback) 0.9.0 protocol features
2|7|APIVERSION|0|Success|[thrd:server1.private:9092/bootstrap]: server1.private/bootstrap: Using (configuration fallback) 0.9.0 protocol features

instead of the FEATURE messages seen in both previous logs

[vin-d]

So looking for PROTOERR and rd_kafka_msgset_reader_msg_v0_1:464 I ended up here #1433 where it's discussed that these PROTOERR are generic normal message in this configuration and that in conclusion it's not an issue. But nobody on that ticket is talking about limited performance per assigned partition.

on another issue about that error https://github.com/confluentinc/confluent-kafka-dotnet/issues/325
you talk about about the Kafka protocol optimisation to use sendfile() syscall.
Not knowing about sendfile I did a small search and found out that there are many saying where it should not be used (for example in nginx conf), there's this related to an os X version https://blog.phusion.nl/2015/06/04/the-brokenness-of-the-sendfile-system-call/
or also this old Apache bug report https://bugs.debian.org/cgi-bin/bugreport.cgi?bug=494768

looking at man sendfile I see the warning

It should not be used in portable programs

VERSIONS
       sendfile() first appeared in Linux 2.2.  The include file <sys/sendfile.h> is present since glibc 2.1.

CONFORMING TO
       Not specified in POSIX.1-2001, nor in other standards.

       Other UNIX systems implement sendfile() with different semantics and prototypes.  It should not be used in portable programs.

So i'm wondering if all this wouldn't be linked to some senfile() issue underneath. I have different Kafka clusters running on Ubuntu 14 and 16, I'll check to see if I get the same results on both.

[vin-d]

Also I don't get this BACKOFF message if I connect to a local Kafka server
would that be because ther's something happening differently on the loopback interface ?
(but I still get the 2 PROTOERR msgs)