engine_dispatch: re-schedule retry when chunk content cannot be read

src/flb_engine_dispatch.c

@@ -90,6 +90,7 @@ int flb_engine_dispatch_retry(struct flb_task_retry *retry,
     char *buf_data;
     size_t buf_size;
     struct flb_task *task;
+    struct flb_output_instance *ins;
 
     task = retry->parent;
 
@@ -116,12 +117,38 @@ int flb_engine_dispatch_retry(struct flb_task_retry *retry,
     /* There is a match, get the buffer */
     buf_data = (char *) flb_input_chunk_flush(task->ic, &buf_size);
     if (!buf_data) {
-        /* Could not retrieve chunk content */
-        flb_error("[engine_dispatch] could not retrieve chunk content, removing retry");
-        record_retry_failure_metrics(task, retry->o_ins, config);
-        flb_task_retry_destroy(retry);
-        flb_task_users_release(task);
-        return -1;
+        /*
+         * The chunk is up but its content could not be read. That is usually
+         * transient, so spend a delivery attempt on it instead of discarding
+         * records a later attempt could still deliver.
+         *
+         * Destroying the retry without releasing the task would leave the task
+         * with no users and no retries, a state nothing reaps.
+         */
+        ins = retry->o_ins;
+
+        if (retry->attempts >= ins->retry_limit && ins->retry_limit >= 0) {
+            flb_error("[engine_dispatch] could not retrieve chunk content, "
+                      "task_id=%i reached retry-attempts limit %i/%i, dropping",
+                      task->id, retry->attempts, ins->retry_limit);
+            record_retry_failure_metrics(task, ins, config);
+            flb_task_retry_destroy(retry);
+            flb_task_users_release(task);
+            return -1;
+        }
+
+        retry->attempts++;
+        flb_warn("[engine_dispatch] could not retrieve chunk content, "
+                 "re-scheduling task_id=%i attempts=%i",
+                 task->id, retry->attempts);
+
+        ret = flb_task_retry_reschedule(retry, config);
+        if (ret == -1) {
+            return -1;
+        }
+
+        /* Just return because it has been re-scheduled */
+        return 0;
     }
 
     /* Update the buffer reference */

tests/internal/engine_dispatch.c

@@ -518,9 +518,83 @@ static void test_retry_flush_failure_preserves_pending_retry(void)
     test_ctx_destroy(ctx);
 }
 
+/*
+ * With retry budget left the chunk must be kept and the attempt re-scheduled,
+ * otherwise a transient read failure would delete records a later attempt
+ * could still deliver.
+ */
+static void test_retry_flush_failure_reschedules_within_retry_limit(void)
+{
+    int ret;
+    int task_id;
+    double value;
+    char *chunk_buffer;
+    char *output_name;
+    struct test_ctx *ctx;
+    struct flb_task *task;
+    struct flb_task_retry *retry;
+    struct flb_output_instance output;
+
+    ctx = test_ctx_create();
+    TEST_CHECK(ctx != NULL);
+    if (ctx == NULL) {
+        return;
+    }
+
+    ret = test_output_init(&output, "output_a");
+    TEST_CHECK(ret == 0);
+    if (ret != 0) {
+        test_ctx_destroy(ctx);
+        return;
+    }
+
+    /* the retry starts with attempts=1, so this leaves budget available */
+    output.retry_limit = 5;
+
+    retry = create_retry_dispatch_task(ctx, &output, &task_id, &chunk_buffer);
+    TEST_CHECK(retry != NULL);
+    if (retry == NULL) {
+        test_output_destroy(&output);
+        test_ctx_destroy(ctx);
+        return;
+    }
+    task = retry->parent;
+
+    ret = flb_engine_dispatch_retry(retry, ctx->config);
+    TEST_CHECK(ret == 0);
+
+    /* the task keeps its slot, its chunk and a pending retry */
+    TEST_CHECK(ctx->config->task_map[task_id].task == task);
+    TEST_CHECK(mk_list_size(&ctx->input->tasks) == 1);
+    TEST_CHECK(mk_list_size(&ctx->input->chunks) == 1);
+    TEST_CHECK(mk_list_size(&task->retries) == 1);
+    TEST_CHECK(retry->attempts == 2);
+
+    /*
+     * Nothing was dropped, so no drop accounting must be recorded. An untouched
+     * counter has no series yet, so cmt_counter_get_val() reports a failure.
+     */
+    output_name = (char *) flb_output_name(&output);
+    ret = cmt_counter_get_val(output.cmt_retries_failed,
+                              1, (char *[]) {output_name}, &value);
+    TEST_CHECK(ret != 0 || value == 0);
+
+    ret = cmt_counter_get_val(output.cmt_dropped_records,
+                              1, (char *[]) {output_name}, &value);
+    TEST_CHECK(ret != 0 || value == 0);
+
+    flb_task_destroy(task, FLB_TRUE);
+    free(chunk_buffer);
+
+    test_output_destroy(&output);
+    test_ctx_destroy(ctx);
+}
+
 TEST_LIST = {
     { "retry_flush_failure_releases_last_task_owner",
       test_retry_flush_failure_releases_last_task_owner },
+    { "retry_flush_failure_reschedules_within_retry_limit",
+      test_retry_flush_failure_reschedules_within_retry_limit },
     { "retry_flush_failure_preserves_active_task_owner",
       test_retry_flush_failure_preserves_active_task_owner },
     { "retry_flush_failure_preserves_pending_retry",