Add recover_failure_document processor to restore failurestore docs to original form

[seanzatzdev]

In our docs at https://www.elastic.co/docs/manage-data/data-store/data-streams/failure-store-recipes#create-a-pipeline-to-convert-failure-documents we currently describe how to remediate documents from the failure store using a Painless script.

To make things less error-prone, this PR encapsulates this functionality into a processor.

e.g.

POST localhost:9200/_ingest/pipeline/_simulate
{
    "pipeline": {
        "processors": [
            {
                "recover_failure_document": {}
            }
        ]
    },
    "docs": [
        {
            "_index": ".fs-my-datastream-ingest-2025.05.09-000001",
            "_id": "HnTJs5YBwrYNjPmaFcri",
            "_score": 1,
            "_source": {
                "@timestamp": "2025-05-09T06:41:24.775Z",
                "document": {
                    "index": "my-datastream-ingest",
                    "source": {
                        "@timestamp": "2025-04-21T00:00:00Z",
                        "counter_name": "test"
                    }
                },
                "error": {
                    "type": "illegal_argument_exception",
                    "message": "field [counter] not present as part of path [counter]",
                    "stack_trace": "j.l.IllegalArgumentException: field [counter] not present as part of path [counter] at o.e.i.IngestDocument.getFieldValue(IngestDocument.java: 202 at o.e.i.c.SetProcessor.execute(SetProcessor.java: 86) 14 more",
                    "pipeline_trace": [
                        "complicated-processor"
                    ],
                    "pipeline": "complicated-processor",
                    "processor_type": "set",
                    "processor_tag": "copy to new counter again"
                }
            }
        }
    ]
}

returns the following response:

{
    "docs": [
        {
            "doc": {
                "_index": "my-datastream-ingest",
                "_version": "-3",
                "_id": "HnTJs5YBwrYNjPmaFcri",
                "_source": {
                    "@timestamp": "2025-04-21T00:00:00Z",
                    "counter_name": "test"
                },
                "_ingest": {
                    "pre_recovery": {
                        "@timestamp": "2025-05-09T06:41:24.775Z",
                        "_index": ".fs-my-datastream-ingest-2025.05.09-000001",
                        "document": {
                            "index": "my-datastream-ingest"
                        },
                        "_id": "HnTJs5YBwrYNjPmaFcri",
                        "error": {
                            "pipeline": "complicated-processor",
                            "processor_type": "set",
                            "processor_tag": "copy to new counter again",
                            "pipeline_trace": [
                                "complicated-processor"
                            ],
                            "stack_trace": "j.l.IllegalArgumentException: field [counter] not present as part of path [counter] at o.e.i.IngestDocument.getFieldValue(IngestDocument.java: 202 at o.e.i.c.SetProcessor.execute(SetProcessor.java: 86) 14 more",
                            "type": "illegal_argument_exception",
                            "message": "field [counter] not present as part of path [counter]"
                        },
                        "_version": -3
                    },
                    "timestamp": "2025-09-04T22:32:12.800709Z"
                }
            }
        }
    ]
}

Using this processor for an invalid failure-doc results in an error, e.g.:

POST localhost:9200/_ingest/pipeline/_simulate

{
    "pipeline": {
        "processors": [
            {
                "recover_failure_document": {}
            }
        ]
    },
    "docs": [
        {
            "_index": ".fs-my-datastream-ingest-2025.05.09-000001",
            "_id": "HnTJs5YBwrYNjPmaFcri",
            "_score": 1,
            "_source": {
                "@timestamp": "2025-05-09T06:41:24.775Z",
                "error": {
                    "type": "illegal_argument_exception",
                    "message": "field [counter] not present as part of path [counter]",
                    "stack_trace": "j.l.IllegalArgumentException: field [counter] not present as part of path [counter] at o.e.i.IngestDocument.getFieldValue(IngestDocument.java: 202 at o.e.i.c.SetProcessor.execute(SetProcessor.java: 86) 14 more",
                    "pipeline_trace": [
                        "complicated-processor"
                    ],
                    "pipeline": "complicated-processor",
                    "processor_type": "set",
                    "processor_tag": "copy to new counter again"
                }
            }
        }
    ]
}

returns

{
    "docs": [
        {
            "error": {
                "root_cause": [
                    {
                        "type": "illegal_argument_exception",
                        "reason": "field [document] not present as part of path [document]"
                    }
                ],
                "type": "illegal_argument_exception",
                "reason": "field [document] not present as part of path [document]"
            }
        }
    ]
}

Closes #132940

[github-actions]

🔍 Preview links for changed docs

• docs/reference/enrich-processor/recover-failure-document-processor.md
• docs/reference/enrich-processor/index.md

[elasticsearchmachine]

Pinging @elastic/es-data-management (Team:Data Management)

[elasticsearchmachine]

Hi @seanzatzdev, I've created a changelog YAML for you.

[github-actions]

ℹ️ Important: Docs version tagging

👋 Thanks for updating the docs! Just a friendly reminder that our docs are now cumulative. This means all 9.x versions are documented on the same page and published off of the main branch, instead of creating separate pages for each minor version.

We use applies_to tags to mark version-specific features and changes.

Expand for a quick overview

When to use applies_to tags:

✅ At the page level to indicate which products/deployments the content applies to (mandatory)
✅ When features change state (e.g. preview, ga) in a specific version
✅ When availability differs across deployments and environments

What NOT to do:

❌ Don't remove or replace information that applies to an older version
❌ Don't add new information that applies to a specific version without an applies_to tag
❌ Don't forget that applies_to tags can be used at the page, section, and inline level

🤔 Need help?

• Check out the cumulative docs guidelines
• Reach out in the #docs Slack channel

[jbaiera]

Left a couple of ideas!

[samxbr]

Maybe it's just me, I find the processor name remediate a bit confusing. This processor itself transforms failure doc to its original doc, but there are more to actually "remediate" the failure doc outside of this processor like reindexing. I wonder if this processor should be named something more accurate to what it does. To throw some ideas, something like failure_source, failure_to_origin?

[seanzatzdev]

Thanks for the feedback @jbaiera and @samxbr , I've updated my PR accordingly

[samxbr]

The PR generally LGTM, I want to leave the final approval to @jbaiera since he is the expert on failure store.

[jbaiera]

This is looking really great! I left just a few small comments, but once those are handled I think it LGTM!

[seanzatzdev]

@jbaiera thanks for the great feedback!