out_kafka: Crash(SIGSEGV) on Kafka producer initialization failure #11358
Description
Bug Report
Describe the bug
When rd_kafka_new() fails during Kafka output plugin initialization, a SIGSEGV crash occurs in flb_out_kafka_destroy().
The crash happens because ctx->topics list is not initialized before calling flb_out_kafka_destroy(). The mk_list_init(&ctx->topics) is called after rd_kafka_new(), so when rd_kafka_new() fails and triggers cleanup, flb_kafka_topic_destroy_all() attempts to iterate over an uninitialized list, causing a segmentation fault.
To Reproduce
Steps to reproduce the problem:
- Configure out_kafka plugin with invalid or unreachable broker settings
Example configuration that can trigger this:
[SERVICE]
flush 1
grace 60
log_level info
parsers_file /tmp/testing/parsers.conf
plugins_file /tmp/testing/plugins.conf
[INPUT]
Name dummy
Tag testing
Dummy {"message": "dummy log"}
Rate 1
[OUTPUT]
name kafka
match *
Brokers 127.0.0.1:9092
Topics testing_topic
timestamp_format iso8601
rdkafka.enable.idempotence true
rdkafka.max.in.flight.requests.per.connection 10000
- Start Fluent Bit
cd build
cmake -DFLB_SIMD=Yes -DFLB_OUT_KAFKA=On -DFLB_RELEASE=On -DFLB_DEBUG=On -DFLB_JEMALLOC=On ..
make
bin/fluent-bit -v -c ./fluentbit.conf
Fluent Bit v5.0.0
* Copyright (C) 2015-2025 The Fluent Bit Authors
* Fluent Bit is a CNCF graduated project under the Fluent organization
* https://fluentbit.io
______ _ _ ______ _ _ _____ _____ _
| ___| | | | | ___ (_) | | ___|| _ | | |
| |_ | |_ _ ___ _ __ | |_ | |_/ /_| |_ __ _|___ \ | |/' |______ __| | _____ __
| _| | | | | |/ _ \ '_ \| __| | ___ \ | __| \ \ / / \ \| /| |______/ _` |/ _ \ \ / /
| | | | |_| | __/ | | | |_ | |_/ / | |_ \ V //\__/ /\ |_/ / | (_| | __/\ V /
\_| |_|\__,_|\___|_| |_|\__| \____/|_|\__| \_/ \____(_)\___/ \__,_|\___| \_/
[2026/01/09 14:43:57.572656604] [ info] [fluent bit] version=5.0.0, commit=e19e07e8f1, pid=414183
[2026/01/09 14:43:57.572713890] [ info] [storage] ver=1.5.4, type=memory, sync=normal, checksum=off, max_chunks_up=128
[2026/01/09 14:43:57.572716207] [ info] [simd ] SSE2
[2026/01/09 14:43:57.572717998] [ info] [cmetrics] version=1.0.6
[2026/01/09 14:43:57.572719473] [ info] [ctraces ] version=0.6.6
[2026/01/09 14:43:57.572756561] [ info] [input:dummy:dummy.0] initializing
[2026/01/09 14:43:57.572758819] [ info] [input:dummy:dummy.0] storage_strategy='memory' (memory only)
[2026/01/09 14:43:57.573352286] [error] [output:kafka:kafka.0] failed to create producer: `max.in.flight` must be set <= 5 when `enable.idempotence` is true
[2026/01/09 14:43:57] [engine] caught signal (SIGSEGV)
#0 0x64e7fd69b2f6 in flb_out_kafka_destroy() at plugins/out_kafka/kafka_config.c:302
#1 0x64e7fd69b9e3 in flb_out_kafka_create() at plugins/out_kafka/kafka_config.c:243
#2 0x64e7fd69995c in cb_kafka_init() at plugins/out_kafka/kafka.c:82
#3 0x64e7fd56a7d3 in flb_output_init_all() at src/flb_output.c:1564
#4 0x64e7fd57e009 in flb_engine_start() at src/flb_engine.c:972
#5 0x64e7fd551e77 in flb_lib_worker() at src/flb_lib.c:904
#6 0x7c2440a9caa3 in start_thread() at nix/sysv/linux/rseq-internal.h:447
#7 0x7c2440b29c6b in clone3() at $|:78
#8 0xffffffffffffffff in ???() at ???:0
Aborted (core dumped)
- rd_kafka_new() fails due to configuration error
- SIGSEGV occurs during cleanup
Expected behavior
When rd_kafka_new() fails, Fluent Bit should gracefully handle the error and clean up resources without crashing. An error message should be logged and the plugin initialization should fail safely.
Screenshots
Your Environment
Ubuntu 24.04
- Version used:
5.0.0 - Configuration:
[SERVICE]
flush 1
grace 60
log_level info
parsers_file /tmp/testing/parsers.conf
plugins_file /tmp/testing/plugins.conf
[INPUT]
Name dummy
Tag testing
Dummy {"message": "dummy log"}
Rate 1
[OUTPUT]
name kafka
match *
Brokers 127.0.0.1:9092
Topics testing_topic
timestamp_format iso8601
rdkafka.enable.idempotence true
rdkafka.max.in.flight.requests.per.connection 10000
- Environment name and version (e.g. Kubernetes? What version?):
- Server type and version:
- Operating System and version:
- Filters and plugins:
Additional context
N/A