build: test branch, ignore

Signed-off-by: Leonardo Alminana <[email protected]>

include/fluent-bit/flb_task.h

@@ -125,6 +125,7 @@ struct flb_task_enqueued {
 
 int flb_task_running_count(struct flb_config *config);
 int flb_task_running_print(struct flb_config *config);
+int flb_task_map_get_task_id(struct flb_config *config);
 
 struct flb_task *flb_task_create(uint64_t ref_id,
                                  const char *buf,

src/flb_engine_dispatch.c

@@ -29,7 +29,7 @@
 #include <fluent-bit/flb_engine.h>
 #include <fluent-bit/flb_task.h>
 #include <fluent-bit/flb_event.h>
-
+#include <chunkio/chunkio.h>
 
 /* It creates a new output thread using a 'Retry' context */
 int flb_engine_dispatch_retry(struct flb_task_retry *retry,
@@ -270,6 +270,17 @@ int flb_engine_dispatch(uint64_t id, struct flb_input_instance *in,
             continue;
         }
 
+        if (flb_task_map_get_task_id(config) == -1) {
+            break;
+        }
+        if (cio_chunk_is_locked(ic->chunk) == CIO_TRUE && flb_task_map_get_task_id(config) == -1) {
+            /*
+             * There isn't a task available and the chunk has been processed
+             * before.
+             */
+            continue;
+        }
+
         /* There is a match, get the buffer */
         buf_data = flb_input_chunk_flush(ic, &buf_size);
         if (buf_size == 0) {
@@ -312,6 +323,12 @@ int flb_engine_dispatch(uint64_t id, struct flb_input_instance *in,
              */
             if (t_err == FLB_TRUE) {
                 flb_input_chunk_release_lock(ic);
+
+                /*
+                 * If the Storage type is 'filesystem' we need to put
+                 * the file content down.
+                 */
+                flb_input_chunk_down(ic);
             }
             continue;
         }

src/flb_task.c

@@ -331,6 +331,10 @@ int flb_task_running_print(struct flb_config *config)
     return 0;
 }
 
+int flb_task_map_get_task_id(struct flb_config *config) {
+    return map_get_task_id(config);
+}
+
 /* Create an engine task to handle the output plugin flushing work */
 struct flb_task *flb_task_create(uint64_t ref_id,
                                  const char *buf,