This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 62049d40f0d NIFI-16247 - Add ingest pipeline support to
PutElasticsearchJson This closes #11589
62049d40f0d is described below
commit 62049d40f0d30f7e64fd088bd80091298ca0901a
Author: Adam Turley <[email protected]>
AuthorDate: Mon Aug 24 14:01:33 2026 -0700
NIFI-16247 - Add ingest pipeline support to PutElasticsearchJson
This closes #11589
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 +++
.../elasticsearch/PutElasticsearchJsonTest.java | 224 +++++++++++++++++++++
3 files changed, 379 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 269f4df77a7..7d34996a296 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 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
.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 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
static final List<PropertyDescriptor> DESCRIPTORS = List.of(
INDEX_OP,
INDEX,
+ PIPELINE,
TYPE,
SCRIPT,
SCRIPTED_UPSERT,
@@ -368,6 +408,8 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
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 class PutElasticsearchJson extends
AbstractPutElasticsearch {
: 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 class PutElasticsearchJson extends
AbstractPutElasticsearch {
|| 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 class PutElasticsearchJson extends
AbstractPutElasticsearch {
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 class PutElasticsearchJson extends
AbstractPutElasticsearch {
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<String, Object> 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 class PutElasticsearchJson extends
AbstractPutElasticsearch {
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 class PutElasticsearchJson extends
AbstractPutElasticsearch {
.script(scriptMap)
.scriptedUpsert(scriptedUpsert)
.dynamicTemplates(dynamicTemplatesMap)
- .headerFields(bulkHeaderFields)
+
.headerFields(withPipeline(bulkHeaderFields, docPipeline))
.build();
docBytes = rawJsonBytes.length;
} else {
final Map<String, Object> 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 class PutElasticsearchJson extends
AbstractPutElasticsearch {
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 class PutElasticsearchJson extends
AbstractPutElasticsearch {
final Map<String, Object> 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 class PutElasticsearchJson extends
AbstractPutElasticsearch {
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 class PutElasticsearchJson extends
AbstractPutElasticsearch {
.script(scriptMap)
.scriptedUpsert(scriptedUpsert)
.dynamicTemplates(dynamicTemplatesMap)
- .headerFields(bulkHeaderFields)
+
.headerFields(withPipeline(bulkHeaderFields, docPipeline))
.build();
chunkBytes += docBytes;
totalBytesAccumulated += docBytes;
@@ -726,13 +794,18 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
final Map<String, Object> 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 class PutElasticsearchJson extends
AbstractPutElasticsearch {
try (final InputStream in = session.read(flowFile)) {
final Map<String, Object> 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 class PutElasticsearchJson extends
AbstractPutElasticsearch {
.script(scriptMap)
.scriptedUpsert(scriptedUpsert)
.dynamicTemplates(dynamicTemplatesMap)
- .headerFields(bulkHeaderFields)
+ .headerFields(withPipeline(bulkHeaderFields,
docPipeline))
.build();
operations.add(opRequest);
operationFlowFiles.add(flowFile);
@@ -1181,31 +1262,45 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
}
/**
- * 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<String, Object> 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<String, Object> contentMap, final
String indexField, final String fallbackIndex) {
- if (StringUtils.isBlank(indexField)) {
- return fallbackIndex;
+ private static Map<String, String> withPipeline(final Map<String, String>
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<String, String> 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 d2da2f79302..21bf715e589 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 c61efe132fb..81af48f2df6 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,227 @@ public class PutElasticsearchJsonTest extends
AbstractPutElasticsearchTest {
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<IndexOperationRequest> 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<IndexOperationRequest> 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<IndexOperationRequest> 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<IndexOperationRequest> 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<IndexOperationRequest> 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<IndexOperationRequest> 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<IndexOperationRequest> 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<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"msg\":\"x\"}");
+ runner.run();
+
+ assertEquals("single-p",
ops.getFirst().getHeaderFields().get("pipeline"));
+ }
+
+ @Test
+ void testDifferentPipelinesPerDocumentInSingleBulk() {
+ // Two NDJSON lines with different Pipeline Field values must be
routed through different ingest
+ // pipelines within the same _bulk request, with the static Pipeline
used where the field is absent.
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.PIPELINE, "default-pipeline");
+ runner.setProperty(PutElasticsearchJson.PIPELINE_FIELD, "pipeline");
+ runner.setProperty(PutElasticsearchJson.RETAIN_PIPELINE_FIELD,
"false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"pipeline\":\"pipeline-a\",\"msg\":\"first\"}\n"
+ + "{\"pipeline\":\"pipeline-b\",\"msg\":\"second\"}\n"
+ + "{\"msg\":\"third\"}\n");
+ runner.run();
+
+ runner.assertTransferCount(PutElasticsearchJson.REL_SUCCESSFUL, 1);
+ assertEquals(3, ops.size());
+ assertEquals("pipeline-a",
ops.get(0).getHeaderFields().get("pipeline"));
+ assertEquals("pipeline-b",
ops.get(1).getHeaderFields().get("pipeline"));
+ assertEquals("default-pipeline",
ops.get(2).getHeaderFields().get("pipeline"));
+ assertFalse(docContent(ops.get(0)).contains("\"pipeline\""));
+ assertFalse(docContent(ops.get(1)).contains("\"pipeline\""));
+ assertTrue(docContent(ops.get(0)).contains("first"));
+ assertTrue(docContent(ops.get(1)).contains("second"));
+ assertTrue(docContent(ops.get(2)).contains("third"));
+ }
+
+ @Test
+ void testPipelineAppliedToCreateOperation() {
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.INDEX_OP,
IndexOperationRequest.Operation.Create.getValue().toLowerCase());
+ runner.setProperty(PutElasticsearchJson.IDENTIFIER_FIELD, "doc_id");
+ runner.setProperty(PutElasticsearchJson.PIPELINE, "default-pipeline");
+ runner.setProperty(PutElasticsearchJson.PIPELINE_FIELD, "pipeline");
+ runner.setProperty(PutElasticsearchJson.RETAIN_PIPELINE_FIELD,
"false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+
runner.enqueue("{\"doc_id\":\"1\",\"pipeline\":\"create-p\",\"msg\":\"hello\"}\n"
+ + "{\"doc_id\":\"2\",\"msg\":\"fallback\"}\n");
+ runner.run();
+
+ runner.assertTransferCount(PutElasticsearchJson.REL_SUCCESSFUL, 1);
+ assertEquals(2, ops.size());
+ assertEquals(IndexOperationRequest.Operation.Create,
ops.get(0).getOperation());
+ assertEquals("create-p", ops.get(0).getHeaderFields().get("pipeline"));
+ assertFalse(docContent(ops.get(0)).contains("\"pipeline\""), "pipeline
field stripped from body");
+ assertEquals(IndexOperationRequest.Operation.Create,
ops.get(1).getOperation());
+ assertEquals("default-pipeline",
ops.get(1).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<IndexOperationRequest> 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<IndexOperationRequest> 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<IndexOperationRequest> 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"));
+ }
}
