input: apply back pressure if input threaded plugin is paused

[danlenar]

Fixes only second scenario described in #7846

If a threaded input plugin is paused due to memory buf limits, dont read chunks/records from the ring buffer.
This will prevent this debug message from being logged.

[2023/08/21 20:35:35] [ info] [task]   task_id=36 still running on route(s): kafka/kafka.0
[2023/08/21 19:55:57] [debug] [input chunk] tail.0 is paused, cannot append records
[2023/08/21 19:55:57] [debug] [input chunk] tail.0 is paused, cannot append records
[2023/08/21 19:55:57] [debug] [input chunk] tail.0 is paused, cannot append records
[2023/08/21 19:55:57] [debug] [input chunk] tail.0 is paused, cannot append records
[2023/08/21 19:55:57] [debug] [input chunk] tail.0 is paused, cannot append records
[2023/08/21 19:55:57] [debug] [input chunk] tail.0 is paused, cannot append records
[2023/08/21 19:55:57] [debug] [input chunk] tail.0 is paused, cannot append records
[2023/08/21 19:55:57] [debug] [input chunk] tail.0 is paused, cannot append records

The fix should replicate similar back pressure as done in non-threaded input plugins.

---

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
• Debug log output from testing the change

• 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

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]

I think that due to the decoupled nature of threaded input plugins we should actually drain the ring buffer regardless of the plugin instance state but I understand that this would be deeper change so this might be fine at least for the moment.

[danlenar]

The other option is to remove this check for threaded plugins if you always want to drain the ring buffer regardless.
https://github.com/fluent/fluent-bit/blob/v2.1.8/src/flb_input_chunk.c#L1430

The only side effect is that mem_buf_limit then tends to have little impact for threaded input plugins.

Also, the triggering of that buffer is paused check in input_chunk_append_raw should be treated as a bug.
If back pressure is working correctly, then there shouldn't be a need to drop records due to a plugin being paused.
A properly paused plugin would have prevented input_chunk_append_raw from being triggered.