Error handling in ElasticIO write method

[damccorm]

ElasticIO write method could sometimes raise Elastic internal errors, like invalid mapping, too many request or other. In this case, it can be usefull to retrieve uncommitted documents into output PCollection to isolate them. What the best way to achieve this ? Could it be implemented directly into ElasticIO connector ?

Imported from Jira BEAM-11309. Original Jira may contain additional context.
Reported by: lionelsetan.

[egalpin]

This is now supported (as of 2.38.0, but strongly recommend using >=2.41.0) by doing the following:

PCollectionTuple esWriteResults = myPcollection.apply(ElasticsearchIO.write().withThrowWriteErrors(false));

PCollection<Document> failedDocs = esWriteResults.get(ElasticsearchIO.Write.FAILED_WRITES);
failedDocs.apply("do something with failed writes", ...)