kafka output plugin does not handle librdkafka fatal errors

[takafumi-kataoka]

Bug Report

Describe the bug

When one of the kafka brokers to which the kafka output plugin was sent went down, the following error was output and subsequent messages were lost.

Failed to produce to topic topic-name: Local: Fatal error

To Reproduce

A related issue says the following

• Set rdkafka.enable.idempotence to true
• Trigger a fatal error. You can do this by killing one of the brokers managing the output topic. This is not guaranteed to trigger a fatal error, but it will succeed eventually.

I can reproduce this in my production environment, but I have not been able to successfully trigger a fatal error in a separate environment.
I will post a method later once I am able to reproduce it.

Expected behavior

Messages are not lost even if a fatal error occurs.

Screenshots

Your Environment

• fluent-bit version: v2.2.1
• OS: Ubuntu 22.04
• Filters and plugins: kafka output plugin

Config

[SERVICE]
    Flush                     1
    Grace                     300
    Log_Level                 info
    Parsers_File ./parsers.conf
    HTTP_Server               On
    Health_Check              On
    HC_Errors_Count           1
    HC_Retry_Failure_Count    1
    HC_Period                 60
    storage.path              /srv/fluent-bit/
    storage.metrics           On

[INPUT]
    Name              tail
    Tag               xxx
    Alias             xxx-input
    Path              /var/log/xxx/*.log
    DB                /var/lib/fluent-bit/xxx.sqlite
    Read_from_Head    True
    Rotate_Wait       180
    storage.type      filesystem
    Mem_Buf_Limit     1mb
    Buffer_Chunk_Size 128kb
    Buffer_Max_Size   128kb
    Parser json

[INPUT]
    Name              tail
    Tag               yyy
    Alias             yyy-input
    Path              /var/log/yyy/*.log
    DB                /var/lib/fluent-bit/yyy.sqlite
    Read_from_Head    True
    Rotate_Wait       180
    storage.type      filesystem
    Mem_Buf_Limit     30mb
    Buffer_Chunk_Size 128kb
    Buffer_Max_Size   128kb
    Parser json

[OUTPUT]
    Name                    kafka
    Match                   xxx
    Brokers                 a.b.c.d:9092,a.b.c.d:9092,a.b.c.d:9092,a.b.c.d:9092,a.b.c.d:9092,a.b.c.d:9092,a.b.c.d:9092
    Topics                  xxx
    Format                  json
    Alias                   xxx-output
    Retry_Limit             False
    rdkafka.acks all
    rdkafka.batch.size 512000
    rdkafka.compression.codec lz4
    rdkafka.enable.idempotence True
    rdkafka.linger.ms 1
    rdkafka.message.send.max.retries 2147483647
    rdkafka.request.timeout.ms 300000
    rdkafka.transaction.timeout.ms 10800000
    rdkafka.socket.timeout.ms 300000
    rdkafka.message.timeout.ms 0

[OUTPUT]
    Name                    kafka
    Match                   yyy
    Brokers                 a.b.c.d:9092,a.b.c.d:9092,a.b.c.d:9092,a.b.c.d:9092,a.b.c.d:9092,a.b.c.d:9092,a.b.c.d:9092
    Topics                  yyy
    Workers                 8
    Format                  json
    Alias                   yyy-output
    Retry_Limit             False
    rdkafka.acks all
    rdkafka.batch.size 512000
    rdkafka.compression.codec lz4
    rdkafka.enable.idempotence True
    rdkafka.linger.ms 1
    rdkafka.message.send.max.retries 2147483647
    rdkafka.request.timeout.ms 300000
    rdkafka.transaction.timeout.ms 10800000
    rdkafka.socket.timeout.ms 300000
    rdkafka.message.timeout.ms 0

Additional context

Reading the librdkafka documentation, it seems that when idempotence is enabled, a fatal error can occur and it is necessary to deal with the error, but it seems that the fluent-bit kafka output plugin continues to send messages even if a fatal error occurs.

I suspect that subsequent messages are also lost.

The fluent-bit version in my environment is v2.2.1, but looking at the code, I think the latest version is the same.