in_kafka: Fix in kafka json parsing

[tarruda]

This fixes json parsing for in_kafka, also add data_format configuration option, for explicit parsing control.

Fixes #7481

[leonardo-albertovich]

We should add this new setting to the documentation, especially if it changes the default behavior considering that before this plugin tried to decode json encoded data always (which is not the case anymore).

[edsiper]

All good with the changes; just one minor ask, please change the config property from data_format to format since we use it in other places.

[tarruda]

All good with the changes; just one minor ask, please change the config property from data_format to format since we use it in other places.

renamed the property @edsiper

[lecaros]

@tarruda where's the doc for the new format option? What are the options for this setting?

[tarruda]

@tarruda where's the doc for the new format option?

fluent/fluent-bit-docs#1118 I just renamed data_format to format in the docs PR.

What are the options for this setting?

For now it is just "none" (no parsing, the default) and "json". Eventually we might support other formats such as avro

[martinmajtan]

Hello,
I tried format option in my configuration and still I got this message:

[2023/07/21 12:01:04] [error] [config] kafka: unknown configuration property 'format'. The following properties are allowed: poll_ms, topics, brokers, client_id, group_id, and rdkafka..

I use this conf:

config:

  inputs: |
    [INPUT]
        Name kafka
        Brokers xxx
        Topics my.topic
        poll_ms 100
        format json

Edit1: Neither format or data_format is not working

[tarruda]

@martinmajtan did you compile fluent-bit from latest master? I just ran the kafka_filter example and it is working for me.

[martinmajtan]

@martinmajtan did you compile fluent-bit from latest master? I just ran the kafka_filter example and it is working for me.

I should have mention I'm using Helm chart where I used image version 2.1.7.

[tarruda]

@martinmajtan this PR will only be available in version 2.1.8

[anosulchik]

[2023/08/15 16:15:42] [ info] [fluent bit] version=2.1.8, commit=1d83649441, pid=1
[2023/08/15 16:15:42] [ info] [storage] ver=1.4.0, type=memory, sync=normal, checksum=off, max_chunks_up=128
[2023/08/15 16:15:42] [ info] [cmetrics] version=0.6.3
[2023/08/15 16:15:42] [ info] [ctraces ] version=0.3.1
[2023/08/15 16:15:42] [ info] [input:kafka:kafka.0] initializing
[2023/08/15 16:15:42] [ info] [input:kafka:kafka.0] storage_strategy='memory' (memory only)
[2023/08/15 16:15:42] [ info] [input:fluentbit_metrics:fluentbit_metrics.1] initializing
[2023/08/15 16:15:42] [ info] [input:fluentbit_metrics:fluentbit_metrics.1] storage_strategy='memory' (memory only)
[2023/08/15 16:15:42] [error] [config] loki: unknown configuration property 'format'. The following properties are allowed: tenant_id, tenant_id_key, labels, auto_kubernetes_labels, drop_single_key, label_keys, remove_keys, line_format, label_map_path, http_user, http_passwd, and bearer_token.
[2023/08/15 16:15:42] [ help] try the command: /fluent-bit/bin/fluent-bit -o loki -h