in_kafka: use log encoder

[Syn3rman]

Use log_encoder instead of mpack in in_kafka

#7301

---

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
	topics quickstart-events
	brokers 0.0.0.0:9092

[OUTPUT]
	Name stdout
	Match *

• Debug log output from testing the change

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

Fluent Bit v2.1.3
* Copyright (C) 2015-2022 The Fluent Bit Authors
* Fluent Bit is a CNCF sub-project under the umbrella of Fluentd
* https://fluentbit.io/

[2023/05/11 08:43:35] [ info] [fluent bit] version=2.1.3, commit=1e62c1b861, pid=45090
[2023/05/11 08:43:35] [ info] [storage] ver=1.4.0, type=memory, sync=normal, checksum=off, max_chunks_up=128
[2023/05/11 08:43:35] [ info] [cmetrics] version=0.6.1
[2023/05/11 08:43:35] [ info] [ctraces ] version=0.3.0
[2023/05/11 08:43:35] [ info] [input:kafka:kafka.0] initializing
[2023/05/11 08:43:35] [ info] [input:kafka:kafka.0] storage_strategy='memory' (memory only)
[2023/05/11 08:43:35] [ info] [sp] stream processor started
[2023/05/11 08:43:35] [ info] [output:stdout:stdout.0] worker #0 started
[0] kafka.0: [[1683794628.892560263, {}], {"topic"=>"quickstart-events", "partition"=>88494920, "offset"=>88494960, "error"=>nil, "key"=>nil, "payload"=>"test1"}]
[0] kafka.0: [[1683794630.378107708, {}], {"topic"=>"quickstart-events", "partition"=>89114664, "offset"=>89114704, "error"=>nil, "key"=>nil, "payload"=>"test2"}]
[2023/05/11 08:43:56] [engine] caught signal (SIGTERM)
[2023/05/11 08:43:57] [ warn] [engine] service will shutdown in max 5 seconds
[2023/05/11 08:43:57] [ info] [engine] service has stopped (0 pending tasks)
[2023/05/11 08:43:58] [ info] [output:stdout:stdout.0] thread worker #0 stopping...
[2023/05/11 08:43:58] [ info] [output:stdout:stdout.0] thread worker #0 stopped
==45090==
==45090== HEAP SUMMARY:
==45090==     in use at exit: 8,033 bytes in 19 blocks
==45090==   total heap usage: 2,600 allocs, 2,581 frees, 2,207,704 bytes allocated
==45090==
==45090== LEAK SUMMARY:
==45090==    definitely lost: 0 bytes in 0 blocks
==45090==    indirectly lost: 0 bytes in 0 blocks
==45090==      possibly lost: 0 bytes in 0 blocks
==45090==    still reachable: 8,033 bytes in 19 blocks
==45090==         suppressed: 0 bytes in 0 blocks
==45090== Reachable blocks (those to which a pointer was found) are not shown.
==45090== To see them, rerun with: --leak-check=full --show-leak-kinds=all
==45090==
==45090== For lists of detected and suppressed errors, rerun with: -s
==45090== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 0 from 0)

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

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.

[leonardo-albertovich]

Once these changes are addressed this will be ready to be merged.