engine: added retry cleanup when a task cannot be retried

Due to how flb_task_users_release determines if it can drop the involved chunk and how entries in the retries list are handled by flb_task_retry_create we need to explicitly remove the retry entries related to the failed task and plugin instance before calling flb_task_users_dec to ensure that chunks linked to tasks that had at least one retry but are deemed not retriable afterwards can be properly disposed of. Signed-off-by: Leonardo Alminana <[email protected]>
fluent · Feb 20, 2023 · 712e5fb · 712e5fb
1 parent 4ce6fb3
commit 712e5fb
Showing 1 changed file with 8 additions and 0 deletions.
diff --git a/src/flb_engine.c b/src/flb_engine.c
@@ -269,6 +269,7 @@ static inline int handle_output_event(flb_pipefd_t fd, uint64_t ts,
                      flb_input_name(task->i_ins),
                      flb_output_name(ins), out_id);
         }
+
         flb_task_retry_clean(task, ins);
         flb_task_users_dec(task, FLB_TRUE);
     }
@@ -288,7 +289,10 @@ static inline int handle_output_event(flb_pipefd_t fd, uint64_t ts,
                      task_id,
                      flb_input_name(task->i_ins),
                      flb_output_name(ins), out_id);
+
+            flb_task_retry_clean(task, ins);
             flb_task_users_dec(task, FLB_TRUE);
+
             return 0;
         }
 
@@ -320,7 +324,9 @@ static inline int handle_output_event(flb_pipefd_t fd, uint64_t ts,
                      flb_input_name(task->i_ins),
                      flb_output_name(ins));
 
+            flb_task_retry_clean(task, ins);
             flb_task_users_dec(task, FLB_TRUE);
+
             return 0;
         }
 
@@ -379,6 +385,8 @@ static inline int handle_output_event(flb_pipefd_t fd, uint64_t ts,
         flb_metrics_sum(FLB_METRIC_OUT_ERROR, 1, ins->metrics);
         flb_metrics_sum(FLB_METRIC_OUT_DROPPED_RECORDS, task->records, ins->metrics);
 #endif
+
+        flb_task_retry_clean(task, ins);
         flb_task_users_dec(task, FLB_TRUE);
     }