in_kafka: Make back pressure workable and add a mechanism of polling threshold

[cosmo0920]

We need to register pause/resume callback for correct handling of back pressure in in_kafka.
Back pressure implementation of Fluent Bit is relying on pause/resume callback which is registered in the each of plugins.
Also, the current implementation of in_kafka does not consider the threshold to halt polling after reaching the limit of chunks.
It causes always over limit status of chunks. This shouldn't happen if the parameters are correctly set up.
I implemented restrictive polling in in_kafka. This shouldn't cause spikes of memory consumption.

---

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
        rdkafka.auto.offset.reset earliest
        # To increase flow rate, specify larger value into buffer_max_size
        # buffer_max_size 5M

• Debug log output from testing the change
  The above config is effective to restrict polling records at once:

<snip>
[2023/11/15 19:01:56] [debug] [input:kafka:kafka.0] kafka message received
[2023/11/15 19:01:56] [debug] [input:kafka:kafka.0] kafka message received
[2023/11/15 19:01:56] [debug] [input:kafka:kafka.0] kafka message received
[2023/11/15 19:01:56] [debug] [input chunk] update output instances with new chunk size diff=5000544, records=2328, input=kafka.0
[2023/11/15 19:01:56] [ warn] [input] kafka.0 paused (mem buf overlimit)
[2023/11/15 19:01:56] [ info] [input] pausing kafka.0
[2023/11/15 19:01:57] [debug] [task] created task=0x7f5db8071fc0 id=0 OK
[2023/11/15 19:01:57] [debug] [output:splunk:splunk.0] task_id=0 assigned to thread #1
[2023/11/15 19:01:57] [debug] [upstream] KA connection #73 to 127.0.0.1:8443 has been assigned (recycled)
[2023/11/15 19:01:57] [debug] [http_client] not using http_proxy for header
[2023/11/15 19:01:57] [debug] [upstream] KA connection #73 to 127.0.0.1:8443 is now available
[2023/11/15 19:01:57] [debug] [out flush] cb_destroy coro_id=20
[2023/11/15 19:01:57] [debug] [task] destroy task=0x7f5db8071fc0 (task_id=0)
[2023/11/15 19:01:57] [ info] [input] kafka.0 resume (mem buf overlimit)
<snip>

Also, when specifying buffer_chunk_size, this parameter is effective to restrict polling events at once:

<snip>
[2023/11/15 19:05:05] [debug] [input:kafka:kafka.0] kafka message received
[2023/11/15 19:05:05] [debug] [input:kafka:kafka.0] kafka message received
[2023/11/15 19:05:05] [debug] [input:kafka:kafka.0] kafka message received
[2023/11/15 19:05:05] [debug] [input:kafka:kafka.0] kafka message received
[2023/11/15 19:05:05] [debug] [input:kafka:kafka.0] kafka message received
[2023/11/15 19:05:05] [debug] [input:kafka:kafka.0] kafka message received
[2023/11/15 19:05:05] [debug] [input:kafka:kafka.0] kafka message received
[2023/11/15 19:05:05] [debug] [input:kafka:kafka.0] kafka message received
[2023/11/15 19:05:05] [debug] [input:kafka:kafka.0] kafka message received
[2023/11/15 19:05:05] [debug] [input:kafka:kafka.0] kafka message received
[2023/11/15 19:05:05] [debug] [input chunk] update output instances with new chunk size diff=5017600, records=2328, input=kafka.0
[2023/11/15 19:05:05] [debug] [task] destroy task=0x7faa6407f2e0 (task_id=0)
<snip>

• 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

fluent/fluent-bit-docs#1257

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.