Possible Memory Leak with Kafka source

[biuboombiuboom]

A note for the community

• Please vote on this issue by adding a 👍 reaction to the original issue to help the community and maintainers prioritize this request
• If you are interested in working on this issue or have submitted a pull request, please leave a comment

Problem

We have a configuration with a kafka source and an clickhouse sink.
The memory often overflow。

Configuration

[sources.in]
session_timeout_ms = 180000
type = "kafka"
bootstrap_servers = "bootstrap.brokers.kafka:9092"
group_id = "kafka-consume-group"
topics = ["my-topic"]
librdkafka_options = { "auto.offset.reset" = "latest", "fetch.min.bytes" = "10240" }

[transforms.balance]
type = "remap"
inputs = ["in"]
drop_on_error = true
source = '.route_id=random_int(0,4)'

[transforms.route]
type = "route"
inputs = ["balance"]
reroute_unmatched = false
route.first = '.route_id==0'
route.second = '.route_id==1'
route.third = '.route_id==2'
route.fouth = '.route_id==3'


[sinks.out-clickhouse-1]
type = "clickhouse"
inputs = ["route.first"]
compression = "none"
endpoint = "http://localhost:8125"
database = "mytable"

batch.max_events = 800000
batch.max_bytes = 2147483648
batch.timeout_secs = 120

request.timeout_secs = 120
request.adaptive_concurrency.initial_concurrency=4
# request.adaptive_concurrency.max_concurrency_limit=4

buffer.type = "memory"
buffer.max_events = 400000
buffer.when_full = "block"

[sinks.out-clickhouse-2]
type = "clickhouse"
inputs = ["route.second"]
compression = "none"
endpoint = "http://localhost:8125"
database = "apmmytable

batch.max_events = 800000
batch.max_bytes = 2147483648
batch.timeout_secs = 120

request.timeout_secs = 120
request.adaptive_concurrency.initial_concurrency=4
# request.adaptive_concurrency.max_concurrency_limit=4

buffer.type = "memory"
buffer.max_events = 400000
buffer.when_full = "block"

[sinks.out-clickhouse-3]
type = "clickhouse"
inputs = ["route.third"]
compression = "none"
endpoint = "http://localhost:8125"
database = "mytable"

batch.max_events = 800000
batch.max_bytes = 2147483648
batch.timeout_secs = 120

request.timeout_secs = 120
request.adaptive_concurrency.initial_concurrency=4
# request.adaptive_concurrency.max_concurrency_limit=4

buffer.type = "memory"
buffer.max_events = 400000
buffer.when_full = "block"

[sinks.out-clickhouse-4]
type = "clickhouse"
inputs = ["route.fouth"]
compression = "none"
endpoint = "http://localhost:8125"
database = "mytable"

batch.max_events = 800000
batch.max_bytes = 2147483648
batch.timeout_secs = 120

request.timeout_secs = 120
request.adaptive_concurrency.initial_concurrency=4
# request.adaptive_concurrency.max_concurrency_limit=4

buffer.type = "memory"
buffer.max_events = 400000
buffer.when_full = "block"



[sources.vector_logs]
type = "internal_logs"

[sinks.log2file]
type = "file"
inputs = ["vector_logs"]
path = "/data/logs/vector-%Y-%m-%d.log"
encoding.codec = "logfmt"

[sources.vector_metrics]
type = "internal_metrics"

[sinks.prometheus_exporter]
inputs = ["vector_metrics"]
type = "prometheus_exporter"

Version

0.37.1

Debug Output

No response

Example Data

No response

Additional Context

No response

References

No response

[jszwedko]

Hi @biuboombiuboom ,

There's not a lot of detail here. What leads you to believe there is a memory leak? If you are feeling motivated you could try running Vector under valgrind and providing the profile here.

[nsweeting]

We're actually seeing a similar issue running the latest version. With a kafka source and a clickhouse sink. A sustained flow of messages seems to inevitably result in OOM. You can see the behaviour in these charts. We have very bursty workloads. Each workload burst results in sustained CPU use (makes sense) as well as continued memory growth. If the work stops - the memory goes back down. But if the work continues... it will OOM.

[biuboombiuboom]

ack_stream has too many FinalizerEntry not consumed.

Try sync store offsets,and disable BatchNotifier.

Or move ack_stream.next() before messages.next() & with biased.

[bruceg]

I'm not clear what you mean by the following:

Try sync store offsets,and disable BatchNotifier.

The second suggestion is spot on and definitely worth a try. The ack_stream should always be checked before consuming new events, because eventually it will run dry to allow progress for new data.

[biuboombiuboom]

I'm not clear what you mean by the following:

Try sync store offsets,and disable BatchNotifier.

The second suggestion is spot on and definitely worth a try. The ack_stream should always be checked before consuming new events, because eventually it will run dry to allow progress for new data.

Sorry ,I didn't express it clearly, please disregard the first suggestion

[bruceg]

I have pushed up a possible fix for this issue. If you are able to build Vector from source, check out PR #20467 and try the resulting binary.

[bruceg]

The proposed fix is now merged into the master branch. Please confirm if this resolves the memory leak for you.

[biuboombiuboom]

The proposed fix is now merged into the master branch. Please confirm if this resolves the memory leak for you.

Yes, It is effective

[jszwedko]

Thanks for confirming! And for suggesting the original fix. I'll close this issue out.