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")); + } }