input: apply back pressure if input threaded plugin is paused (#7847)

Signed-off-by: Daniel Lenar <[email protected]>
fluent · Sep 4, 2023 · 9cb07ca · 9cb07ca
1 parent 6578ccc
commit 9cb07ca
Showing 1 changed file with 12 additions and 3 deletions.
diff --git a/src/flb_input_chunk.c b/src/flb_input_chunk.c
@@ -1766,9 +1766,18 @@ void flb_input_chunk_ring_buffer_collector(struct flb_config *ctx, void *data)
         ins = mk_list_entry(head, struct flb_input_instance, _head);
         cr = NULL;
 
-        while ((ret = flb_ring_buffer_read(ins->rb,
-                                           (void *) &cr,
-                                           sizeof(cr))) == 0) {
+        while (1) {
+            if (flb_input_buf_paused(ins) == FLB_TRUE) {
+                break;
+            }
+
+            ret = flb_ring_buffer_read(ins->rb,
+                                       (void *) &cr,
+                                       sizeof(cr));
+            if (ret != 0) {
+                break;
+            }
+
             if (cr) {
                 if (cr->tag) {
                     tag_len = flb_sds_len(cr->tag);