From 75a78b2e0de0afe260de8e57a94d76a811e5f925 Mon Sep 17 00:00:00 2001 From: Adam Turley Date: Mon, 24 Aug 2026 14:01:33 -0700 Subject: [PATCH] NIFI-16247 - Add ingest pipeline support to PutElasticsearchJson Add Pipeline, Pipeline Field, and Retain Pipeline Field properties, mirroring the existing Index configuration. Pipeline sets a static ingest pipeline for every document, while Pipeline Field resolves the pipeline per document from the payload and honors the Field Path Mode property, so the value can come from a top-level field or a nested "/"-delimited path. When the field is absent or the property is blank, the static Pipeline value is used as the fallback. The resolved value is written to the bulk action header for Index and Create operations, so records within a single FlowFile can be routed through different ingest pipelines in one _bulk request. Retain Pipeline Field controls whether the field is stripped from the document body before indexing; the field is stripped whenever requested, including for operations the pipeline itself does not apply to, so routing metadata is never indexed. Generalize the index name extraction helpers to resolveFieldValue and extractFieldValue since they are now shared with the pipeline resolution. --- .../elasticsearch/PutElasticsearchJson.java | 155 +++++++++++++--- .../additionalDetails.md | 30 +++ .../PutElasticsearchJsonTest.java | 172 ++++++++++++++++++ 3 files changed, 327 insertions(+), 30 deletions(-) diff --git a/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java b/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java index 269f4df77a79..7d34996a296a 100644 --- a/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java +++ b/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java @@ -333,6 +333,45 @@ For a nested ("/"-delimited) path, any parent object left empty by the removal i .dependsOn(TIMESTAMP_FIELD) .build(); + static final PropertyDescriptor PIPELINE = new PropertyDescriptor.Builder() + .name("Pipeline") + .description(""" + The name of the Elasticsearch ingest pipeline to run the documents through. \ + Applies to Index and Create operations only. When left blank, no pipeline is set unless \ + provided per-document by the Pipeline Field property.\ + """) + .required(false) + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + static final PropertyDescriptor PIPELINE_FIELD = new PropertyDescriptor.Builder() + .name("Pipeline Field") + .description(""" + The name of the field within each document to use as the Elasticsearch ingest pipeline, \ + interpreted as a literal field name or a nested "/"-delimited path per the Field Path Mode property. \ + Applies to Index and Create operations only. If the field is not present in a document or this \ + property is left blank, the configured Pipeline property value is used as the fallback.\ + """) + .required(false) + .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .build(); + + static final PropertyDescriptor RETAIN_PIPELINE_FIELD = new PropertyDescriptor.Builder() + .name("Retain Pipeline Field") + .description(""" + Whether to keep the Pipeline Field in the document body after extracting it \ + for use as the Elasticsearch ingest pipeline. \ + When true (default), the field is left in the document; set to false to remove it before indexing. \ + For a nested ("/"-delimited) path, any parent object left empty by the removal is also pruned.\ + """) + .required(true) + .allowableValues("true", "false") + .defaultValue("true") + .dependsOn(PIPELINE_FIELD) + .build(); + static final Relationship REL_BULK_REQUEST = new Relationship.Builder() .name("bulk_request") .description("When \"Output Bulk Request\" is enabled, the raw Elasticsearch _bulk API request body is written " + @@ -352,6 +391,7 @@ For a nested ("/"-delimited) path, any parent object left empty by the removal i static final List DESCRIPTORS = List.of( INDEX_OP, INDEX, + PIPELINE, TYPE, SCRIPT, SCRIPTED_UPSERT, @@ -368,6 +408,8 @@ For a nested ("/"-delimited) path, any parent object left empty by the removal i RETAIN_INDEX_FIELD, TIMESTAMP_FIELD, RETAIN_TIMESTAMP_FIELD, + PIPELINE_FIELD, + RETAIN_PIPELINE_FIELD, CHARSET, MAX_JSON_FIELD_STRING_LENGTH, CLIENT_SERVICE, @@ -496,6 +538,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session : null; final String documentIndexField = fieldPath(context.getProperty(INDEX_FIELD).evaluateAttributeExpressions().getValue(), nestedFieldPaths); final String documentTimestampField = fieldPath(context.getProperty(TIMESTAMP_FIELD).evaluateAttributeExpressions().getValue(), nestedFieldPaths); + final String documentPipelineField = fieldPath(context.getProperty(PIPELINE_FIELD).evaluateAttributeExpressions().getValue(), nestedFieldPaths); // The id/index field paths are loop-invariant, so classify them as nested-vs-flat and decode // the flat names once here rather than per record in the NDJSON raw-bytes fast path below. final boolean nestedExtractionField = isNestedPath(documentIdField) || isNestedPath(documentIndexField); @@ -511,6 +554,8 @@ public void onTrigger(final ProcessContext context, final ProcessSession session || context.getProperty(RETAIN_INDEX_FIELD).asBoolean(); final boolean retainTimestampField = StringUtils.isBlank(documentTimestampField) || context.getProperty(RETAIN_TIMESTAMP_FIELD).asBoolean(); + final boolean retainPipelineField = StringUtils.isBlank(documentPipelineField) + || context.getProperty(RETAIN_PIPELINE_FIELD).asBoolean(); final int batchSize = InputFormat.SINGLE_JSON == inputFormat ? context.getProperty(BATCH_SIZE).evaluateAttributeExpressions().asInteger() : Integer.MAX_VALUE; @@ -536,6 +581,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session while (flowFile != null) { final String indexOp = context.getProperty(INDEX_OP).evaluateAttributeExpressions(flowFile).getValue(); final String index = context.getProperty(INDEX).evaluateAttributeExpressions(flowFile).getValue(); + final String pipeline = context.getProperty(PIPELINE).evaluateAttributeExpressions(flowFile).getValue(); final String type = context.getProperty(TYPE).evaluateAttributeExpressions(flowFile).getValue(); final String charset = context.getProperty(CHARSET).evaluateAttributeExpressions(flowFile).getValue(); final String flowFileIdAttribute = StringUtils.isNotBlank(idAttribute) ? flowFile.getAttribute(idAttribute) : null; @@ -568,22 +614,29 @@ public void onTrigger(final ProcessContext context, final ProcessSession session final byte[] rawJsonBytes; final String id; final String docIndex; + final String docPipeline; final boolean stripId = !retainIdentifierField && StringUtils.isNotBlank(documentIdField); final boolean stripIdx = !retainIndexField && StringUtils.isNotBlank(documentIndexField); + final boolean stripPipeline = !retainPipelineField && StringUtils.isNotBlank(documentPipelineField); final boolean needsTimestamp = StringUtils.isNotBlank(documentTimestampField); - // The raw streaming scan only matches flat field names; a nested - // (/-delimited) id or index path requires the parsed Map. - if (suppressingWriter != null || stripId || stripIdx || needsTimestamp || nestedExtractionField) { - // Map is needed anyway — extract both fields from the Map directly. + final boolean needsPipelineFromField = StringUtils.isNotBlank(documentPipelineField); + // The raw streaming scan only matches flat field names; a nested (/-delimited) id or + // index path, or a pipeline read from the payload, requires the parsed Map. + if (suppressingWriter != null || stripId || stripIdx || needsTimestamp || nestedExtractionField || needsPipelineFromField) { + // Map is needed anyway — extract the fields from the Map directly. final Map contentMap = mapReader.readValue(trimmedLine); id = resolveId(contentMap, documentIdField, flowFileIdAttribute); - docIndex = resolveIndex(contentMap, documentIndexField, index); + docIndex = resolveFieldValue(contentMap, documentIndexField, index); + docPipeline = resolveFieldValue(contentMap, documentPipelineField, pipeline); if (stripId) { removeAtPath(contentMap, documentIdField); } if (stripIdx) { removeAtPath(contentMap, documentIndexField); } + if (stripPipeline) { + removeAtPath(contentMap, documentPipelineField); + } applyTimestamp(contentMap, documentTimestampField, retainTimestampField); rawJsonBytes = suppressingWriter != null ? suppressingWriter.writeValueAsBytes(contentMap) @@ -594,6 +647,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session final String[] extracted = extractIdAndIndex(trimmedLine, documentIdFieldName, flowFileIdAttribute, documentIndexFieldName, index); id = extracted[0]; docIndex = extracted[1]; + docPipeline = pipeline; rawJsonBytes = trimmedLine.getBytes(StandardCharsets.UTF_8); } opRequest = IndexOperationRequest.builder() @@ -605,19 +659,24 @@ public void onTrigger(final ProcessContext context, final ProcessSession session .script(scriptMap) .scriptedUpsert(scriptedUpsert) .dynamicTemplates(dynamicTemplatesMap) - .headerFields(bulkHeaderFields) + .headerFields(withPipeline(bulkHeaderFields, docPipeline)) .build(); docBytes = rawJsonBytes.length; } else { final Map contentMap = mapReader.readValue(trimmedLine); final String id = resolveId(contentMap, documentIdField, flowFileIdAttribute); - final String docIndex = resolveIndex(contentMap, documentIndexField, index); + final String docIndex = resolveFieldValue(contentMap, documentIndexField, index); if (!retainIdentifierField && StringUtils.isNotBlank(documentIdField)) { removeAtPath(contentMap, documentIdField); } if (!retainIndexField && StringUtils.isNotBlank(documentIndexField)) { removeAtPath(contentMap, documentIndexField); } + // The pipeline is not applied to this operation, but the field is still stripped when + // requested so the routing metadata is not indexed with the document. + if (!retainPipelineField && StringUtils.isNotBlank(documentPipelineField)) { + removeAtPath(contentMap, documentPipelineField); + } applyTimestamp(contentMap, documentTimestampField, retainTimestampField); opRequest = IndexOperationRequest.builder() .index(docIndex) @@ -673,6 +732,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session final byte[] rawJsonBytes; final String id; final String docIndex; + final String docPipeline; if (suppressingWriter != null) { // Parse directly to Map so NON_NULL/NON_EMPTY inclusion filters apply during // serialization. JsonNode tree serialization bypasses JsonInclude filters, @@ -680,20 +740,25 @@ public void onTrigger(final ProcessContext context, final ProcessSession session final Map contentMap = mapReader.readValue(parser); docBytes = Math.max(1, parser.currentLocation().getCharOffset() - startOffset); id = resolveId(contentMap, documentIdField, flowFileIdAttribute); - docIndex = resolveIndex(contentMap, documentIndexField, index); + docIndex = resolveFieldValue(contentMap, documentIndexField, index); + docPipeline = resolveFieldValue(contentMap, documentPipelineField, pipeline); if (!retainIdentifierField && StringUtils.isNotBlank(documentIdField)) { removeAtPath(contentMap, documentIdField); } if (!retainIndexField && StringUtils.isNotBlank(documentIndexField)) { removeAtPath(contentMap, documentIndexField); } + if (!retainPipelineField && StringUtils.isNotBlank(documentPipelineField)) { + removeAtPath(contentMap, documentPipelineField); + } applyTimestamp(contentMap, documentTimestampField, retainTimestampField); rawJsonBytes = suppressingWriter.writeValueAsBytes(contentMap); } else { final JsonNode node = mapper.readTree(parser); docBytes = Math.max(1, parser.currentLocation().getCharOffset() - startOffset); id = extractId(node, documentIdField, flowFileIdAttribute); - docIndex = extractIndex(node, documentIndexField, index); + docIndex = extractFieldValue(node, documentIndexField, index); + docPipeline = extractFieldValue(node, documentPipelineField, pipeline); // Field stripping and @timestamp injection only apply to JSON objects. // Non-object elements (scalars, arrays, null) are passed through unchanged so // Elasticsearch can reject them per-document rather than failing the whole FlowFile. @@ -705,6 +770,9 @@ public void onTrigger(final ProcessContext context, final ProcessSession session if (!retainIndexField && StringUtils.isNotBlank(documentIndexField)) { removeAtPath(objectNode, documentIndexField); } + if (!retainPipelineField && StringUtils.isNotBlank(documentPipelineField)) { + removeAtPath(objectNode, documentPipelineField); + } applyTimestamp(objectNode, documentTimestampField, retainTimestampField); } rawJsonBytes = mapper.writeValueAsBytes(node); @@ -718,7 +786,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session .script(scriptMap) .scriptedUpsert(scriptedUpsert) .dynamicTemplates(dynamicTemplatesMap) - .headerFields(bulkHeaderFields) + .headerFields(withPipeline(bulkHeaderFields, docPipeline)) .build(); chunkBytes += docBytes; totalBytesAccumulated += docBytes; @@ -726,13 +794,18 @@ public void onTrigger(final ProcessContext context, final ProcessSession session final Map contentMap = mapReader.readValue(parser); final long docBytes = Math.max(1, parser.currentLocation().getCharOffset() - startOffset); final String id = resolveId(contentMap, documentIdField, flowFileIdAttribute); - final String docIndex = resolveIndex(contentMap, documentIndexField, index); + final String docIndex = resolveFieldValue(contentMap, documentIndexField, index); if (!retainIdentifierField && StringUtils.isNotBlank(documentIdField)) { removeAtPath(contentMap, documentIdField); } if (!retainIndexField && StringUtils.isNotBlank(documentIndexField)) { removeAtPath(contentMap, documentIndexField); } + // The pipeline is not applied to this operation, but the field is still stripped + // when requested so the routing metadata is not indexed with the document. + if (!retainPipelineField && StringUtils.isNotBlank(documentPipelineField)) { + removeAtPath(contentMap, documentPipelineField); + } applyTimestamp(contentMap, documentTimestampField, retainTimestampField); opRequest = IndexOperationRequest.builder() .index(docIndex) @@ -767,10 +840,18 @@ public void onTrigger(final ProcessContext context, final ProcessSession session try (final InputStream in = session.read(flowFile)) { final Map contentMap = mapReader.readValue(in); final String id = StringUtils.isNotBlank(flowFileIdAttribute) ? flowFileIdAttribute : null; - final String docIndex = resolveIndex(contentMap, documentIndexField, index); + final String docIndex = resolveFieldValue(contentMap, documentIndexField, index); + // Ingest pipelines apply to Index/Create operations only. + final boolean pipelineApplies = o == IndexOperationRequest.Operation.Index || o == IndexOperationRequest.Operation.Create; + final String docPipeline = pipelineApplies ? resolveFieldValue(contentMap, documentPipelineField, pipeline) : null; if (!retainIndexField && StringUtils.isNotBlank(documentIndexField)) { removeAtPath(contentMap, documentIndexField); } + // The field is stripped when requested even for operations the pipeline does not apply to, + // so the routing metadata is not indexed with the document. + if (!retainPipelineField && StringUtils.isNotBlank(documentPipelineField)) { + removeAtPath(contentMap, documentPipelineField); + } applyTimestamp(contentMap, documentTimestampField, retainTimestampField); final IndexOperationRequest opRequest = IndexOperationRequest.builder() .index(docIndex) @@ -781,7 +862,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session .script(scriptMap) .scriptedUpsert(scriptedUpsert) .dynamicTemplates(dynamicTemplatesMap) - .headerFields(bulkHeaderFields) + .headerFields(withPipeline(bulkHeaderFields, docPipeline)) .build(); operations.add(opRequest); operationFlowFiles.add(flowFile); @@ -1181,31 +1262,45 @@ private String resolveId(final Map contentMap, final String idAt } /** - * Extracts the index name from a pre-parsed {@link JsonNode}. - * Used for JSON Array Index/Create operations where the node is already available. + * Extracts a string value (e.g. index name or ingest pipeline) at {@code field} from a pre-parsed + * {@link JsonNode}. Used for JSON Array Index/Create operations where the node is already available. * The field may be a {@code /}-delimited path into nested objects. - * Falls back to {@code fallbackIndex} when the field is absent or blank. + * Falls back to {@code fallback} when the field is absent or blank. + */ + private String extractFieldValue(final JsonNode node, final String field, final String fallback) { + if (StringUtils.isBlank(field)) { + return fallback; + } + final String value = fieldNodeToString(nodeAtPath(node, field)); + return StringUtils.isNotBlank(value) ? value : fallback; + } + + /** + * Resolves a string value (e.g. index name or ingest pipeline) at {@code field} from an already-parsed + * content Map. The field may be a {@code /}-delimited path into nested objects. + * Falls back to {@code fallback} when the field is absent or blank. */ - private String extractIndex(final JsonNode node, final String indexField, final String fallbackIndex) { - if (StringUtils.isBlank(indexField)) { - return fallbackIndex; + private String resolveFieldValue(final Map contentMap, final String field, final String fallback) { + if (StringUtils.isBlank(field)) { + return fallback; } - final String value = fieldNodeToString(nodeAtPath(node, indexField)); - return StringUtils.isNotBlank(value) ? value : fallbackIndex; + final String value = fieldValueToString(valueAtPath(contentMap, field)); + return StringUtils.isNotBlank(value) ? value : fallback; } /** - * Resolves the index name from an already-parsed content Map. - * Used for Update/Delete/Upsert operations and suppression-enabled Index/Create paths - * where the Map is already available. The field may be a {@code /}-delimited path into - * nested objects. Falls back to {@code fallbackIndex} when the field is absent or blank. + * Returns the bulk action-header map for a document: the shared {@code headerFields} plus a + * {@code pipeline} entry when {@code pipeline} is non-blank (the ingest pipeline is expressed as a + * bulk action-header field). Returns the shared map unchanged when the pipeline is blank so no + * per-document copy is allocated in the common case. */ - private String resolveIndex(final Map contentMap, final String indexField, final String fallbackIndex) { - if (StringUtils.isBlank(indexField)) { - return fallbackIndex; + private static Map withPipeline(final Map headerFields, final String pipeline) { + if (StringUtils.isBlank(pipeline)) { + return headerFields; } - final String value = fieldValueToString(valueAtPath(contentMap, indexField)); - return StringUtils.isNotBlank(value) ? value : fallbackIndex; + final Map merged = new HashMap<>(headerFields); + merged.put("pipeline", pipeline); + return merged; } /** diff --git a/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/docs/org.apache.nifi.processors.elasticsearch.PutElasticsearchJson/additionalDetails.md b/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/docs/org.apache.nifi.processors.elasticsearch.PutElasticsearchJson/additionalDetails.md index d2da2f79302a..21bf715e5895 100644 --- a/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/docs/org.apache.nifi.processors.elasticsearch.PutElasticsearchJson/additionalDetails.md +++ b/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/docs/org.apache.nifi.processors.elasticsearch.PutElasticsearchJson/additionalDetails.md @@ -86,6 +86,36 @@ value is extracted. For a nested path, any parent object that is left empty by t extracting and removing `@metadata/id` from `{"@metadata": {"id": "abc"}, "message": "Hello, world"}` leaves `{"message": "Hello, world"}`, with the now-empty `@metadata` object removed. +### Ingest Pipeline + +Documents can be routed through an [Elasticsearch ingest pipeline](https://www.elastic.co/guide/en/elasticsearch/reference/current/ingest.html) +at index time. This applies to **Index** and **Create** operations only. The pipeline is added to the bulk action +header for each document (`{"index": {"_index": "...", "pipeline": "..."}}`) and can be set two ways: + +* **Pipeline** — a static pipeline name applied to every document (supports Expression Language). +* **Pipeline Field** — the name of a field within each document whose value is the pipeline, resolved per document. + Like the Index Field, this honors the **Field Path Mode** property, so it can be read from a top-level field or a + nested `/`-delimited path. When *Pipeline Field* is blank or absent from a document, the static *Pipeline* property + is used as the fallback. + +**Retain Pipeline Field** controls whether the field is left in the document body or removed before indexing (with +empty parent objects pruned, as described above). + +For example, with *Pipeline Field* set to `@metadata/pipeline` (in Nested Field Path mode) and *Retain Pipeline Field* +set to `false`, the document: + +```json +{ + "@metadata": { + "pipeline": "my-ingest-pipeline" + }, + "message": "Hello, world" +} +``` + +is indexed as `{"message": "Hello, world"}` with the bulk action header `"pipeline": "my-ingest-pipeline"`, so +different documents in the same FlowFile can be routed through different ingest pipelines based on their own content. + ### Dynamic Templates Index and Create operations can use Dynamic Templates. The Dynamic Templates property must be parsable as a JSON object. diff --git a/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJsonTest.java b/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJsonTest.java index c61efe132fbd..e7a07f0df1b7 100644 --- a/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJsonTest.java +++ b/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJsonTest.java @@ -16,6 +16,7 @@ */ package org.apache.nifi.processors.elasticsearch; +import org.apache.nifi.elasticsearch.ElasticSearchClientService; import org.apache.nifi.elasticsearch.IndexOperationRequest; import org.apache.nifi.elasticsearch.IndexOperationResponse; import org.apache.nifi.processor.exception.ProcessException; @@ -1536,4 +1537,175 @@ void testEscapedSlashIdentifierFieldUpdateOperation() { assertFalse(ops.getFirst().getFields().containsKey("a/b"), "escaped-slash flat id field removed"); assertTrue(ops.getFirst().getFields().containsKey("msg")); } + + // ------------------------------------------------------------------------- + // Ingest pipeline (static Pipeline property + per-document Pipeline Field) + // ------------------------------------------------------------------------- + + @Test + void testStaticPipelineAddedToBulkHeader() { + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.PIPELINE, "my-pipeline"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"msg\":\"hello\"}\n"); + runner.run(); + + assertEquals("my-pipeline", ops.getFirst().getHeaderFields().get("pipeline")); + } + + @Test + void testNoPipelineByDefault() { + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"msg\":\"hello\"}\n"); + runner.run(); + + assertFalse(ops.getFirst().getHeaderFields().containsKey("pipeline")); + } + + @Test + void testPipelineFieldFromPayloadRemovedWhenRetainFalse() { + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.PIPELINE_FIELD, "pipeline"); + runner.setProperty(PutElasticsearchJson.RETAIN_PIPELINE_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"pipeline\":\"per-doc\",\"msg\":\"hello\"}\n"); + runner.run(); + + assertEquals("per-doc", ops.getFirst().getHeaderFields().get("pipeline")); + assertFalse(docContent(ops.getFirst()).contains("\"pipeline\""), "pipeline field stripped from body"); + } + + @Test + void testPipelineFieldRetainedByDefault() { + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.PIPELINE_FIELD, "pipeline"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"pipeline\":\"per-doc\",\"msg\":\"hello\"}\n"); + runner.run(); + + assertEquals("per-doc", ops.getFirst().getHeaderFields().get("pipeline")); + assertTrue(docContent(ops.getFirst()).contains("\"pipeline\""), "field retained by default"); + } + + @Test + void testPipelineFieldFallsBackToStaticPipeline() { + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.PIPELINE, "default-pipeline"); + runner.setProperty(PutElasticsearchJson.PIPELINE_FIELD, "pipeline"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"msg\":\"no pipeline field here\"}\n"); + runner.run(); + + assertEquals("default-pipeline", ops.getFirst().getHeaderFields().get("pipeline")); + } + + @Test + void testNestedPipelineFieldPrunesEmptyParent() { + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + useNestedFieldPaths(); + runner.setProperty(PutElasticsearchJson.PIPELINE_FIELD, "@metadata/pipeline"); + runner.setProperty(PutElasticsearchJson.RETAIN_PIPELINE_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"@metadata\":{\"pipeline\":\"nested-p\"},\"msg\":\"hello\"}\n"); + runner.run(); + + assertEquals("nested-p", ops.getFirst().getHeaderFields().get("pipeline")); + assertFalse(docContent(ops.getFirst()).contains("@metadata"), "empty parent pruned after extracting nested pipeline"); + } + + @Test + void testPipelineFieldJsonArray() { + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.JSON_ARRAY.getValue()); + runner.setProperty(PutElasticsearchJson.PIPELINE_FIELD, "pipeline"); + runner.setProperty(PutElasticsearchJson.RETAIN_PIPELINE_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("[{\"pipeline\":\"arr-p\",\"msg\":\"x\"}]"); + runner.run(); + + assertEquals("arr-p", ops.getFirst().getHeaderFields().get("pipeline")); + assertFalse(docContent(ops.getFirst()).contains("\"pipeline\"")); + } + + @Test + void testStaticPipelineSingleJson() { + runner.setProperty(PutElasticsearchJson.PIPELINE, "single-p"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"msg\":\"x\"}"); + runner.run(); + + assertEquals("single-p", ops.getFirst().getHeaderFields().get("pipeline")); + } + + @Test + void testPipelineNotAppliedToUpdateOperation() { + // Ingest pipelines apply to Index/Create only, not Update/Delete/Upsert. + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.INDEX_OP, IndexOperationRequest.Operation.Update.getValue().toLowerCase()); + runner.setProperty(PutElasticsearchJson.IDENTIFIER_FIELD, "doc_id"); + runner.setProperty(PutElasticsearchJson.PIPELINE, "my-pipeline"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"doc_id\":\"1\",\"msg\":\"hello\"}\n"); + runner.run(); + + assertFalse(ops.getFirst().getHeaderFields().containsKey("pipeline"), "pipeline not set for update operations"); + } + + @Test + void testPipelineFieldStrippedEvenWhenPipelineNotApplied() { + // The pipeline is not applied to Update operations, but Retain Pipeline Field = false must still remove + // the field so the routing metadata is not indexed, matching the Index Field behaviour. + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.NDJSON.getValue()); + runner.setProperty(PutElasticsearchJson.INDEX_OP, IndexOperationRequest.Operation.Update.getValue().toLowerCase()); + runner.setProperty(PutElasticsearchJson.IDENTIFIER_FIELD, "doc_id"); + runner.setProperty(PutElasticsearchJson.PIPELINE_FIELD, "pipeline"); + runner.setProperty(PutElasticsearchJson.RETAIN_PIPELINE_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("{\"doc_id\":\"1\",\"pipeline\":\"p\",\"msg\":\"hello\"}\n"); + runner.run(); + + assertFalse(ops.getFirst().getHeaderFields().containsKey("pipeline"), "pipeline not set for update operations"); + assertFalse(ops.getFirst().getFields().containsKey("pipeline"), "pipeline field still stripped from the body"); + assertTrue(ops.getFirst().getFields().containsKey("msg")); + } + + @Test + void testPipelineFieldWithSuppressNulls() { + // The JSON Array suppressing-writer path parses to a Map rather than a JsonNode; verify the + // pipeline is resolved and stripped there too. + runner.setProperty(PutElasticsearchJson.INPUT_FORMAT, InputFormat.JSON_ARRAY.getValue()); + runner.setProperty(PutElasticsearchJson.SUPPRESS_NULLS, ElasticSearchClientService.ALWAYS_SUPPRESS.getValue()); + runner.setProperty(PutElasticsearchJson.PIPELINE_FIELD, "pipeline"); + runner.setProperty(PutElasticsearchJson.RETAIN_PIPELINE_FIELD, "false"); + runner.assertValid(); + + final List ops = captureOperations(); + runner.enqueue("[{\"pipeline\":\"suppress-p\",\"msg\":\"x\",\"nullable\":null}]"); + runner.run(); + + assertEquals("suppress-p", ops.getFirst().getHeaderFields().get("pipeline")); + final String content = docContent(ops.getFirst()); + assertFalse(content.contains("\"pipeline\""), "pipeline field stripped on the suppressing path"); + assertTrue(content.contains("msg")); + } }