in_kafka: Add async commit mode

[HaChanho]

We've added async_commit mode to Fluent Bit's in_kafka plugin, which allows you to handle Kafka commit operations asynchronously. Depending on your choice, this is expected to improve performance and efficient resource usage.

---

Enter [N/A] in the box, if an item is not applicable to your change.

Testing
Before we can approve your change; please submit the following in a comment:

• Example configuration file for the change

[INPUT]
    name kafka
    brokers 127.0.0.1:9092
    topics test
    async_commit true

• Debug log output from testing the change

async_commit mode OFF

<snip>
[2024/05/30 09:23:57] [debug] [input:kafka:kafka.0] kafka message received
[2024/05/30 09:23:57] [debug] [input:kafka:kafka.0] offset committed(sync)
[2024/05/30 09:23:57] [debug] [input:kafka:kafka.0] kafka message received
[2024/05/30 09:23:57] [debug] [input:kafka:kafka.0] offset committed(sync)
[2024/05/30 09:23:57] [debug] [input:kafka:kafka.0] kafka message received
[2024/05/30 09:23:57] [debug] [input:kafka:kafka.0] offset committed(sync)
[2024/05/30 09:23:57] [debug] [input:kafka:kafka.0] kafka message received
[2024/05/30 09:23:57] [debug] [input:kafka:kafka.0] offset committed(sync)
[2024/05/30 09:23:57] [debug] [input:kafka:kafka.0] kafka message received
[2024/05/30 09:23:57] [debug] [input:kafka:kafka.0] offset committed(sync)
[2024/05/30 09:23:57] [debug] [input:kafka:kafka.0] kafka message received
[2024/05/30 09:23:57] [debug] [input:kafka:kafka.0] offset committed(sync)
[2024/05/30 09:23:57] [debug] [input:kafka:kafka.0] kafka message received
[2024/05/30 09:23:57] [debug] [input:kafka:kafka.0] offset committed(sync)
<snip>

async_commit mode ON

[2024/05/30 09:49:19] [debug] [input:kafka:kafka.0] kafka message received
[2024/05/30 09:49:19] [debug] [input:kafka:kafka.0] offset committed(async)
[2024/05/30 09:49:19] [debug] [input:kafka:kafka.0] kafka message received
[2024/05/30 09:49:19] [debug] [input:kafka:kafka.0] offset committed(async)
[2024/05/30 09:49:19] [debug] [input:kafka:kafka.0] kafka message received
[2024/05/30 09:49:19] [debug] [input:kafka:kafka.0] offset committed(async)
[2024/05/30 09:49:19] [debug] [input:kafka:kafka.0] kafka message received
[2024/05/30 09:49:19] [debug] [input:kafka:kafka.0] offset committed(async)
[2024/05/30 09:49:19] [debug] [input:kafka:kafka.0] kafka message received
[2024/05/30 09:49:19] [debug] [input:kafka:kafka.0] offset committed(async)
[2024/05/30 09:49:19] [debug] [input:kafka:kafka.0] kafka message received
[2024/05/30 09:49:19] [debug] [input:kafka:kafka.0] offset committed(async)
[2024/05/30 09:49:19] [debug] [input:kafka:kafka.0] kafka message received
[2024/05/30 09:49:19] [debug] [input:kafka:kafka.0] offset committed(async)

• Attached Valgrind output that shows no leaks or memory corruption was found

If this is a change to packaging of containers or native binaries then please confirm it works for all targets.

• Run local packaging test showing all targets (including any new ones) build.
• Set ok-package-test label to test for all targets (requires maintainer to do).

Documentation

• Documentation required for this feature
  in_kafka: Add async_commit description fluent-bit-docs#1385

Backporting

• Backport to latest stable release.

---

Fluent Bit is licensed under Apache 2.0, by submitting this pull request I understand that this code will be released under the terms of that license.

[edsiper]

@HaChanho thanks for this contribution.

just a side question, what if rd_kafka_commit() is called after the iteration happens, once the while has finished ?, now it runs right after every message which has been consumed:

fluent-bit/plugins/in_kafka/in_kafka.c

Line 184 in ea7b9f4

rd_kafka_commit(ctx->kafka.rk, NULL, 0);

[HaChanho]

@edsiper

just a side question, what if rd_kafka_commit() is called after the iteration happens, once the while has finished ?, now it runs right after every message which has been consumed:

If we process the commit after the loop ends, it will certainly improve performance compared to committing right after consuming each message. However, there is a possibility of duplicate message processing if the consumer restarts before the offset is updated due to commit delay.

I believe that consumer restart is rare under normal circumstances, and handling duplicate messages may not be a significant issue compared to message loss. It might be better to handle duplicate messages in the final processing system (output).

Do you have any plans to improve the code related to this?

[github-actions]

This PR is stale because it has been open 45 days with no activity. Remove stale label or comment or this will be closed in 10 days.