[ML] Individual bulk index failures in data frame should be treated as index failures

[hendrikmuhs]

Follow up from #43194

Steps to reproduce:

1. Create a dataset
2. Create data frame transform
3. Create destination index with conflicting mappings
4. Start data frame transform
5. Notice that it finishes without reporting any indexing failures in the _stats API, even though ES logs show 100% of the docs have failed to index.

Bug:
There are 2 issues:

• the indexer does not fail
• the error count is wrong (we should count the bulk index failure as a whole, not the individual index failures)

Solution:
The error handling should use the same error handler that we use for search failures, meaning it should classify the issue into:

• temporary failures (e.g. individual shard failures due to trouble on the node holding the shard)
• irrecoverable failures (e.g. mapping conflict)

Temporary failures shall be retried, irrecoverable failures will set the state to failed without retry.

Retry will re-create/re-index the full page as a whole but not the individual index requests (which would be very complex).

/cc @dolaru @sophiec20

Closes #43194

[elasticmachine]

Pinging @elastic/ml-core

[sophiec20]

Sounds good.

Please clarify for the benefit of this ticket, what would happen in the event that a lesser percentage of the docs failed to index. (this may be difficult to replicate since #43194, but I would image we will experience it in some form)

[hendrikmuhs]

Please clarify for the benefit of this ticket, what would happen in the event that a lesser percentage of the docs failed to index. (this may be difficult to replicate since #43194, but I would image we will experience it in some form)

"Retry will re-create/re-index the full page as a whole but not the individual index requests"

After a successful query/bulk index we advance a "cursor", query/index again with the new cursor, advance, query/index again, ... and so on, until the cursor reaches the end. In case of a failure the cursor will not be advanced and retry will start from the old cursor position (same happens if a node dies and we migrate to another node).

Due to the way bulk indexing works the successful index requests will result in successfully indexed documents. We will not try to roll back those documents. The retry will query and index the full page, which means writing the previously failed documents and overwriting the successful ones (same content, same doc id).

Example: 8 out of 500 index requests failed: means 492 got indexed. We do not advance the cursor, so we do the exact same query and the exact same bulk index request again, therefore we eventually write 500 docs like in the non-failure case. At lucene level we create 8 new docs (which also could be updates!) and 492 updates with identical content.

Alternatives

only retry the failed documents: this would be fairly complex, we would need to remember the failed docs, maybe persist them. I am not expecting to run into this issue often, because we assume stable clusters with stable network connectivity. Therefore this optimization is not worth it.

accept some loss: I think there is nothing worse than incomplete/wrong data. If there is a usecase for accepting loss (high frequency updates, low-latency) it might be an option to allow this as a setting, my gut feeling: this is not what data frames are made for.

roll-back: Taking the 8:492 example from above: The data frame isn't in a stable state anyway, it's somewhere in the middle of a checkpoint. Neither do the 500 documents form a transaction, therefore it seems unnecessary trying to roll-back the 492 documents as it would not end in a consistent state, but you would need to rollback the whole checkpoint. Roll-back is not delete, for continuous data frame you would require to restore the old document. This again would be fairly complex.