{"record":{"id":"a25d803323a0a679","repo":"elastic/elasticsearch","slug":"failure-store-document-has-unexpected-structure-m-a25d80","errorCode":null,"errorMessage":"failure store document has unexpected structure, missing required [document.source] field","messagePattern":"failure store document has unexpected structure, missing required \\[document\\.source\\] field","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/RecoverFailureDocumentProcessor.java","lineNumber":58,"sourceCode":"    public static final String MISSING_SOURCE_ERROR_MSG =\n        \"failure store document has unexpected structure, missing required [document.source] field\";\n    public static final String MISSING_ERROR_ERROR_MSG = \"failure store document has unexpected structure, missing required [error] field\";\n\n    public static final String TYPE = \"recover_failure_document\";\n\n    RecoverFailureDocumentProcessor(String tag, String description) {\n        super(tag, description);\n    }\n\n    @Override\n    @SuppressWarnings(\"unchecked\")\n    public IngestDocument execute(IngestDocument document) throws Exception {\n        if (document.hasField(DOCUMENT_FIELD) == false) {\n            throw new IllegalArgumentException(MISSING_DOCUMENT_ERROR_MSG);\n        }\n\n        if (document.hasField(SOURCE_FIELD_PATH) == false) {\n            throw new IllegalArgumentException(MISSING_SOURCE_ERROR_MSG);\n        }\n\n        if (document.hasField(ERROR_FIELD) == false) {\n            throw new IllegalArgumentException(MISSING_ERROR_ERROR_MSG);\n        }\n\n        // store pre-recovery data in ingest metadata\n        storePreRecoveryData(document);\n\n        // Get the nested 'document' field, which holds the original document and metadata.\n        Map<String, Object> failedDocument = (Map<String, Object>) document.getFieldValue(DOCUMENT_FIELD, Map.class);\n\n        // Copy the original index, routing, and id back to the document's metadata.\n        String originalIndex = (String) failedDocument.get(INDEX_FIELD);\n        if (originalIndex != null) {\n            document.setFieldValue(IngestDocument.Metadata.INDEX.getFieldName(), originalIndex);\n        }\n","sourceCodeStart":40,"sourceCodeEnd":76,"githubUrl":"https://github.com/elastic/elasticsearch/blob/db6a809a667c081ca1dc7500389d26975573215f/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/RecoverFailureDocumentProcessor.java#L40-L76","documentation":"Thrown by RecoverFailureDocumentProcessor.execute when the incoming IngestDocument does not contain the nested 'document.source' field. This processor reconstructs the original document from a failure-store entry created by the pipeline's on-failure handling, so it expects a specific envelope structure with 'document', 'document.source', and 'error' keys.","triggerScenarios":"Calling the recover_failure_document processor on a document that lacks the 'document.source' field path. This happens when the input document was not produced by the failure store or was manually constructed without the full failure envelope.","commonSituations":"Misconfiguring a pipeline that chains into recover_failure_document without first routing through the failure store. Sending an ad-hoc document to a pipeline containing this processor. Schema drift between failure-store versions.","solutions":["Verify the document is a genuine failure-store entry containing the 'document' map with a nested 'source' key before invoking this processor.","Ensure the pipeline is only used in the failure-store recovery path, not on raw documents.","Inspect the document with a print/simulate pipeline step to confirm the 'document.source' field is present."],"exampleFix":"// before: feeding a raw document to the recover_failure_document processor\n// after: ensure the document is a failure-store entry\n// doc must contain: { \"document\": { \"source\": {...} }, \"error\": {...} }\nif (ingestDocument.hasField(\"document.source\") == false) {\n    throw new IllegalStateException(\"input is not a failure-store entry\");\n}","handlingStrategy":"validation","validationCode":"// Before invoking recover_failure_document, verify the envelope structure\nif (document.hasField(\"document\") == false\n    || document.hasField(\"document.source\") == false) {\n    // skip recovery; the input is not a failure-store entry\n    return;\n}","typeGuard":"boolean isFailureStoreEntry(IngestDocument doc) {\n    return doc.hasField(\"document\")\n        && doc.hasField(\"document.source\")\n        && doc.hasField(\"error\");\n}","tryCatchPattern":"try {\n    // run pipeline with recover_failure_document\n} catch (IllegalArgumentException e) {\n    if (e.getMessage().contains(\"missing required [document.source]\")) {\n        // log and route to a dead-letter index; input is not a failure-store entry\n    } else { throw e; }\n}","preventionTips":["Only chain recover_failure_document after the failure-store write path.","Validate the three required fields (document, document.source, error) in a pre-check processor.","Use the simulate pipeline API to test with representative failure-store entries."],"tags":["elasticsearch","ingest-pipeline","recover-failure-document","validation"],"backgroundTag":null,"analyzedSha":"db6a809a667c081ca1dc7500389d26975573215f","analyzedAt":"2026-08-12T01:39:14.192Z","schemaVersion":2},"datasetVersion":"2026-08-13T09:17:06.757Z"}