Handle item-level failures when doing bulk ingestion

[artem-shelkovnikov]

Problem Description

Connectors contain logic here that collects documents to bulk operations and then sends batches into Elasticsearch.

There is a todo in the code:

# TODO: retry 429s for individual items here

It is possible that the whole request will return 200, but individual items within this request will be marked as failed - e.g. you've sent 500 items and 3 of them returned 429 because EIS or ML node was unable to process them.

So we need to add logic that collects all items that failed ingestion in the batch and retry them a couple times. Actual implementation of this needs to be discusssed.

Proposed Solution

When bulk request returns successful status but individual items are not successful these items need to be retried - either combined with more items from the queue or just these items alone. If after a couple attempts ingestion fails these items should be dropped with a log line.

Additional Context

Be mindful of Error Monitor in _process_bulk_response - it counts failed items and makes a decision to drop some items, this logic should only execute after retrying of the ingestion happened

[mahendra-shah]

Hi @artem-shelkovnikov !
I'd like to work on this issue.

My understanding of the problem:
The _process_bulk_response method in sink.py currently handles bulk-level failures but doesn't retry individual items that fail with a 429 status inside an otherwise successful (HTTP 200) bulk response. The existing TODO comment on line ~203 confirms this gap.

My proposed approach:
After processing the bulk response, collect all items where the individual status code indicates failure (specifically 429)
Implement a retry loop (e.g. 3 attempts with exponential backoff) that re-sends only the failed items
Ensure this retry logic runs before the existing Error Monitor in _process_bulk_response makes its drop decision — so the error count only reflects items that failed after all retries, not before
After exhausting retries, log a warning with item details and drop them

One question before I start: Should the retry apply only to 429 status codes, or also to other transient errors like 503?
The issue mentions 429 specifically but I want to confirm scope before writing code.
I'll set up the dev environment and have a draft PR ready within a few days.

[seanstory]

hey @mahendra-shah , thanks for your interest! Sorry for being slow to respond.

Should the retry apply only to 429 status codes, or also to other transient errors like 503?

I think all transient errors is a good idea. However, for 429's specifically, bonus points if you implement respecting any Retry-After header.

[Shaurya2k06]

Hey @mahendra-shah , are you still working on this? I went over your approach, it seems optimal given the issue scope. Happy to pick up from where you left off on.