This is an automated email from the ASF dual-hosted git repository.
pvillard31 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 6b7fba5f331 NIFI-15985 - Add per-document Index Field and Timestamp
Field extraction to PutElasticsearchJson (#11299)
6b7fba5f331 is described below
commit 6b7fba5f3312c026e17beb2951474a523f81ef59
Author: agturley <[email protected]>
AuthorDate: Tue Jun 2 10:40:47 2026 -0700
NIFI-15985 - Add per-document Index Field and Timestamp Field extraction to
PutElasticsearchJson (#11299)
---
.../elasticsearch/PutElasticsearchJson.java | 331 +++++++++++++++++++--
.../elasticsearch/PutElasticsearchJsonTest.java | 229 ++++++++++++++
2 files changed, 531 insertions(+), 29 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 2b10871da38..29b05d854c1 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
@@ -26,6 +26,7 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectReader;
import com.fasterxml.jackson.databind.ObjectWriter;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.nifi.annotation.behavior.DynamicProperties;
import org.apache.nifi.annotation.behavior.DynamicProperty;
import org.apache.nifi.annotation.behavior.InputRequirement;
@@ -65,6 +66,7 @@ import java.io.InputStreamReader;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -83,7 +85,10 @@ import java.util.concurrent.atomic.AtomicBoolean;
"NDJSON (one JSON object per line), JSON Array (a top-level array of
objects, streamed for memory efficiency), " +
"and Single JSON (the entire FlowFile is one document). " +
"FlowFiles are accumulated up to the configured Max Batch Size and
flushed to Elasticsearch in _bulk API requests. " +
- "Large files that exceed the batch size are automatically split into
multiple _bulk requests.")
+ "Large files that exceed the batch size are automatically split into
multiple _bulk requests. " +
+ "Records routed to the \"successful\" and \"errors\" relationships
reflect the original FlowFile content, " +
+ "not the transformed document body sent to Elasticsearch
(identifier/index/timestamp field extraction and " +
+ "null suppression are not reapplied), so the records remain suitable
for replay.")
@WritesAttributes({
@WritesAttribute(attribute = "elasticsearch.put.error",
description = "The error message if there is an issue parsing
the FlowFile, sending the parsed document to Elasticsearch or parsing the
Elasticsearch response"),
@@ -245,6 +250,69 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
.dependsOn(INPUT_FORMAT, InputFormat.NDJSON,
InputFormat.JSON_ARRAY)
.build();
+ static final PropertyDescriptor RETAIN_IDENTIFIER_FIELD = new
PropertyDescriptor.Builder()
+ .name("Retain Identifier Field")
+ .description("""
+ Whether to keep the Identifier Field in the document body
after extracting it \
+ for use as the Elasticsearch document ID. \
+ When true (default), the field is left in the document;
set to false to remove it before indexing.\
+ """)
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .dependsOn(IDENTIFIER_FIELD)
+ .build();
+
+ static final PropertyDescriptor INDEX_FIELD = new
PropertyDescriptor.Builder()
+ .name("Index Field")
+ .description("""
+ The name of the field within each document to use as the
Elasticsearch index name. \
+ If the field is not present in a document or this property
is left blank, \
+ the configured Index property value is used as the
fallback.\
+ """)
+ .required(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor RETAIN_INDEX_FIELD = new
PropertyDescriptor.Builder()
+ .name("Retain Index Field")
+ .description("""
+ Whether to keep the Index Field in the document body after
extracting it \
+ for use as the Elasticsearch index name. \
+ When true (default), the field is left in the document;
set to false to remove it before indexing.\
+ """)
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .dependsOn(INDEX_FIELD)
+ .build();
+
+ static final PropertyDescriptor TIMESTAMP_FIELD = new
PropertyDescriptor.Builder()
+ .name("Timestamp Field")
+ .description("""
+ The name of a field within each document whose value will
be written to \
+ Elasticsearch as the @timestamp field. \
+ If the field is absent or this property is left blank, no
@timestamp is set.\
+ """)
+ .required(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor RETAIN_TIMESTAMP_FIELD = new
PropertyDescriptor.Builder()
+ .name("Retain Timestamp Field")
+ .description("""
+ Whether to keep the Timestamp Field in the document body
after copying its \
+ value to @timestamp. \
+ When true (default), the field is left in the document;
set to false to remove it before indexing.\
+ """)
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .dependsOn(TIMESTAMP_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 " +
@@ -274,6 +342,11 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
SUPPRESS_NULLS,
ID_ATTRIBUTE,
IDENTIFIER_FIELD,
+ RETAIN_IDENTIFIER_FIELD,
+ INDEX_FIELD,
+ RETAIN_INDEX_FIELD,
+ TIMESTAMP_FIELD,
+ RETAIN_TIMESTAMP_FIELD,
CHARSET,
MAX_JSON_FIELD_STRING_LENGTH,
CLIENT_SERVICE,
@@ -396,6 +469,18 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
final String documentIdField = inputFormat != InputFormat.SINGLE_JSON
?
context.getProperty(IDENTIFIER_FIELD).evaluateAttributeExpressions().getValue()
: null;
+ final String documentIndexField =
context.getProperty(INDEX_FIELD).evaluateAttributeExpressions().getValue();
+ final String documentTimestampField =
context.getProperty(TIMESTAMP_FIELD).evaluateAttributeExpressions().getValue();
+ // The Retain toggles depend on their source field being set, so only
read them when it is —
+ // reading a property whose dependency is unsatisfied is invalid. When
the source field is
+ // blank no extraction or stripping occurs, so the retain value is
irrelevant anyway.
+ final boolean retainIdentifierField = inputFormat !=
InputFormat.SINGLE_JSON
+ && StringUtils.isNotBlank(documentIdField)
+ && context.getProperty(RETAIN_IDENTIFIER_FIELD).asBoolean();
+ final boolean retainIndexField =
StringUtils.isBlank(documentIndexField)
+ || context.getProperty(RETAIN_INDEX_FIELD).asBoolean();
+ final boolean retainTimestampField =
StringUtils.isBlank(documentTimestampField)
+ || context.getProperty(RETAIN_TIMESTAMP_FIELD).asBoolean();
final int batchSize = InputFormat.SINGLE_JSON == inputFormat
?
context.getProperty(BATCH_SIZE).evaluateAttributeExpressions().asInteger()
: Integer.MAX_VALUE;
@@ -450,17 +535,36 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
final IndexOperationRequest opRequest;
final long docBytes;
if (o == IndexOperationRequest.Operation.Index ||
o == IndexOperationRequest.Operation.Create) {
- final String id = extractId(trimmedLine,
documentIdField, flowFileIdAttribute);
final byte[] rawJsonBytes;
- if (suppressingWriter != null) {
- // Parse to Map so NON_NULL/NON_EMPTY
inclusion filters apply during serialization.
- // JsonNode tree serialization bypasses
JsonInclude filters.
- rawJsonBytes =
suppressingWriter.writeValueAsBytes(mapReader.readValue(trimmedLine));
+ final String id;
+ final String docIndex;
+ final boolean stripId = !retainIdentifierField
&& StringUtils.isNotBlank(documentIdField);
+ final boolean stripIdx = !retainIndexField &&
StringUtils.isNotBlank(documentIndexField);
+ final boolean needsTimestamp =
StringUtils.isNotBlank(documentTimestampField);
+ if (suppressingWriter != null || stripId ||
stripIdx || needsTimestamp) {
+ // Map is needed anyway — extract both
fields from the Map directly.
+ final Map<String, Object> contentMap =
mapReader.readValue(trimmedLine);
+ id = resolveId(contentMap,
documentIdField, flowFileIdAttribute);
+ docIndex = resolveIndex(contentMap,
documentIndexField, index);
+ if (stripId) {
+ contentMap.remove(documentIdField);
+ }
+ if (stripIdx) {
+ contentMap.remove(documentIndexField);
+ }
+ applyTimestamp(contentMap,
documentTimestampField, retainTimestampField);
+ rawJsonBytes = suppressingWriter != null
+ ?
suppressingWriter.writeValueAsBytes(contentMap)
+ :
mapper.writeValueAsBytes(contentMap);
} else {
+ // Raw-bytes path: single streaming scan
finds both fields at once.
+ final String[] extracted =
extractIdAndIndex(trimmedLine, documentIdField, flowFileIdAttribute,
documentIndexField, index);
+ id = extracted[0];
+ docIndex = extracted[1];
rawJsonBytes =
trimmedLine.getBytes(StandardCharsets.UTF_8);
}
opRequest = IndexOperationRequest.builder()
- .index(index)
+ .index(docIndex)
.type(type)
.id(id)
.rawJsonBytes(rawJsonBytes)
@@ -474,8 +578,16 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
} else {
final Map<String, Object> contentMap =
mapReader.readValue(trimmedLine);
final String id = resolveId(contentMap,
documentIdField, flowFileIdAttribute);
+ final String docIndex =
resolveIndex(contentMap, documentIndexField, index);
+ if (!retainIdentifierField &&
StringUtils.isNotBlank(documentIdField)) {
+ contentMap.remove(documentIdField);
+ }
+ if (!retainIndexField &&
StringUtils.isNotBlank(documentIndexField)) {
+ contentMap.remove(documentIndexField);
+ }
+ applyTimestamp(contentMap,
documentTimestampField, retainTimestampField);
opRequest = IndexOperationRequest.builder()
- .index(index)
+ .index(docIndex)
.type(type)
.id(id)
.fields(contentMap)
@@ -527,22 +639,45 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
final long docBytes;
final byte[] rawJsonBytes;
final String id;
+ final String docIndex;
if (suppressingWriter != null) {
// Parse directly to Map so
NON_NULL/NON_EMPTY inclusion filters apply during
// serialization. JsonNode tree
serialization bypasses JsonInclude filters,
// and convertValue(node, Map) adds an
extra serialization cycle.
final Map<String, Object> contentMap =
mapReader.readValue(parser);
docBytes = Math.max(1,
parser.currentLocation().getCharOffset() - startOffset);
- rawJsonBytes =
suppressingWriter.writeValueAsBytes(contentMap);
id = resolveId(contentMap,
documentIdField, flowFileIdAttribute);
+ docIndex = resolveIndex(contentMap,
documentIndexField, index);
+ if (!retainIdentifierField &&
StringUtils.isNotBlank(documentIdField)) {
+ contentMap.remove(documentIdField);
+ }
+ if (!retainIndexField &&
StringUtils.isNotBlank(documentIndexField)) {
+
contentMap.remove(documentIndexField);
+ }
+ applyTimestamp(contentMap,
documentTimestampField, retainTimestampField);
+ rawJsonBytes =
suppressingWriter.writeValueAsBytes(contentMap);
} else {
final JsonNode node =
mapper.readTree(parser);
docBytes = Math.max(1,
parser.currentLocation().getCharOffset() - startOffset);
- rawJsonBytes =
mapper.writeValueAsBytes(node);
id = extractId(node, documentIdField,
flowFileIdAttribute);
+ docIndex = extractIndex(node,
documentIndexField, index);
+ // 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.
+ if (node.isObject()) {
+ final ObjectNode objectNode =
(ObjectNode) node;
+ if (!retainIdentifierField &&
StringUtils.isNotBlank(documentIdField)) {
+
objectNode.remove(documentIdField);
+ }
+ if (!retainIndexField &&
StringUtils.isNotBlank(documentIndexField)) {
+
objectNode.remove(documentIndexField);
+ }
+ applyTimestamp(objectNode,
documentTimestampField, retainTimestampField);
+ }
+ rawJsonBytes =
mapper.writeValueAsBytes(node);
}
opRequest = IndexOperationRequest.builder()
- .index(index)
+ .index(docIndex)
.type(type)
.id(id)
.rawJsonBytes(rawJsonBytes)
@@ -558,8 +693,16 @@ 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);
+ if (!retainIdentifierField &&
StringUtils.isNotBlank(documentIdField)) {
+ contentMap.remove(documentIdField);
+ }
+ if (!retainIndexField &&
StringUtils.isNotBlank(documentIndexField)) {
+ contentMap.remove(documentIndexField);
+ }
+ applyTimestamp(contentMap,
documentTimestampField, retainTimestampField);
opRequest = IndexOperationRequest.builder()
- .index(index)
+ .index(docIndex)
.type(type)
.id(id)
.fields(contentMap)
@@ -591,8 +734,13 @@ 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);
+ if (!retainIndexField &&
StringUtils.isNotBlank(documentIndexField)) {
+ contentMap.remove(documentIndexField);
+ }
+ applyTimestamp(contentMap, documentTimestampField,
retainTimestampField);
final IndexOperationRequest opRequest =
IndexOperationRequest.builder()
- .index(index)
+ .index(docIndex)
.type(type)
.id(id)
.fields(contentMap)
@@ -881,24 +1029,85 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
}
/**
- * Extracts the document ID from a raw JSON string using a streaming
parser.
- * Stops as soon as the target field is found, avoiding a full tree parse.
- * Used for Index/Create operations to avoid the Map allocation overhead.
+ * Copies the value of {@code timestampField} to {@code @timestamp} in the
Map.
+ * If {@code retain} is false, the source field is removed after copying.
+ * Does nothing when {@code timestampField} is blank or not present in the
document.
*/
- private String extractId(final String rawJson, final String idAttribute,
final String flowFileIdAttribute) throws IOException {
- if (StringUtils.isBlank(idAttribute)) {
- return StringUtils.isNotBlank(flowFileIdAttribute) ?
flowFileIdAttribute : null;
+ private void applyTimestamp(final Map<String, Object> contentMap, final
String timestampField, final boolean retain) {
+ if (StringUtils.isBlank(timestampField)) {
+ return;
+ }
+ final Object value = contentMap.get(timestampField);
+ if (value != null) {
+ if (!retain) {
+ contentMap.remove(timestampField);
+ }
+ contentMap.put("@timestamp", value);
}
+ }
+
+ /**
+ * Copies the value of {@code timestampField} to {@code @timestamp} in the
ObjectNode.
+ * If {@code retain} is false, the source field is removed after copying.
+ * Does nothing when {@code timestampField} is blank or not present in the
document.
+ */
+ private void applyTimestamp(final ObjectNode node, final String
timestampField, final boolean retain) {
+ if (StringUtils.isBlank(timestampField)) {
+ return;
+ }
+ final JsonNode value = node.get(timestampField);
+ if (value != null && !value.isNull()) {
+ if (!retain) {
+ node.remove(timestampField);
+ }
+ node.set("@timestamp", value);
+ }
+ }
+
+ /**
+ * Extracts both document ID and index name from a raw JSON string in a
single streaming pass.
+ * Stops as soon as both fields have been found to avoid scanning the rest
of the document.
+ * Returns a two-element array: {@code [id, docIndex]}.
+ * Used for NDJSON Index/Create when neither suppression nor field-removal
requires a Map parse.
+ */
+ private String[] extractIdAndIndex(
+ final String rawJson,
+ final String idField, final String flowFileIdAttribute,
+ final String indexField, final String fallbackIndex) throws
IOException {
+ final boolean needId = StringUtils.isNotBlank(idField);
+ final boolean needIndex = StringUtils.isNotBlank(indexField);
+
+ String id = needId ? null :
(StringUtils.isNotBlank(flowFileIdAttribute) ? flowFileIdAttribute : null);
+ String docIndex = fallbackIndex;
+
+ if (!needId && !needIndex) {
+ return new String[]{id, docIndex};
+ }
+
+ boolean foundId = !needId;
+ boolean foundIndex = !needIndex;
+
try (final JsonParser p = mapper.getFactory().createParser(rawJson)) {
while (p.nextToken() != null) {
- if (idAttribute.equals(p.currentName()) && p.nextToken() !=
null && !p.currentToken().isStructStart()) {
+ if (foundId && foundIndex) {
+ break;
+ }
+ if (!foundId && idField.equals(p.currentName()) &&
p.nextToken() != null && isScalarValue(p.currentToken())) {
+ final String value = p.getText();
+ id = StringUtils.isNotBlank(value) ? value :
(StringUtils.isNotBlank(flowFileIdAttribute) ? flowFileIdAttribute : null);
+ foundId = true;
+ } else if (!foundIndex && indexField.equals(p.currentName())
&& p.nextToken() != null && isScalarValue(p.currentToken())) {
final String value = p.getText();
- return StringUtils.isNotBlank(value) ? value
- : (StringUtils.isNotBlank(flowFileIdAttribute) ?
flowFileIdAttribute : null);
+ docIndex = StringUtils.isNotBlank(value) ? value :
fallbackIndex;
+ foundIndex = true;
}
}
}
- return StringUtils.isNotBlank(flowFileIdAttribute) ?
flowFileIdAttribute : null;
+
+ if (!foundId) {
+ id = StringUtils.isNotBlank(flowFileIdAttribute) ?
flowFileIdAttribute : null;
+ }
+ return new String[]{id, docIndex};
}
/**
@@ -909,9 +1118,9 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
if (StringUtils.isBlank(idAttribute)) {
return StringUtils.isNotBlank(flowFileIdAttribute) ?
flowFileIdAttribute : null;
}
- final JsonNode idNode = node.get(idAttribute);
- if (idNode != null && !idNode.isNull()) {
- return idNode.asText();
+ final String value = fieldNodeToString(node.get(idAttribute));
+ if (StringUtils.isNotBlank(value)) {
+ return value;
}
return StringUtils.isNotBlank(flowFileIdAttribute) ?
flowFileIdAttribute : null;
}
@@ -924,13 +1133,77 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
if (StringUtils.isBlank(idAttribute)) {
return null;
}
- final Object idObj = contentMap.get(idAttribute);
- if (idObj != null) {
- return idObj.toString();
+ final String value = fieldValueToString(contentMap.get(idAttribute));
+ if (StringUtils.isNotBlank(value)) {
+ return value;
}
return StringUtils.isNotBlank(flowFileIdAttribute) ?
flowFileIdAttribute : null;
}
+ /**
+ * Extracts the index name from a pre-parsed {@link JsonNode}.
+ * Used for JSON Array Index/Create operations where the node is already
available.
+ * Falls back to {@code fallbackIndex} 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;
+ }
+ final String value = fieldNodeToString(node.get(indexField));
+ return StringUtils.isNotBlank(value) ? value : fallbackIndex;
+ }
+
+ /**
+ * 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. Falls back to {@code fallbackIndex}
when the
+ * field is absent or blank.
+ */
+ private String resolveIndex(final Map<String, Object> contentMap, final
String indexField, final String fallbackIndex) {
+ if (StringUtils.isBlank(indexField)) {
+ return fallbackIndex;
+ }
+ final String value = fieldValueToString(contentMap.get(indexField));
+ return StringUtils.isNotBlank(value) ? value : fallbackIndex;
+ }
+
+ /**
+ * Converts a document field value taken from a parsed content Map into
its string form for use
+ * as a document ID or index name. Returns {@code null} for absent values
and for container
+ * values (JSON objects/arrays, deserialized as {@link Map}/{@link
Collection}), which are not
+ * meaningful as an identifier or index and are treated the same as an
absent field. This mirrors
+ * {@link #fieldNodeToString(JsonNode)} and the streaming {@link
#extractIdAndIndex} path so all
+ * three produce identical results for the same document.
+ */
+ private static String fieldValueToString(final Object value) {
+ if (value == null || value instanceof Map || value instanceof
Collection) {
+ return null;
+ }
+ return value.toString();
+ }
+
+ /**
+ * Converts a document field {@link JsonNode} into its string form for use
as a document ID or
+ * index name. Returns {@code null} for missing, null, and non-scalar
(object/array) nodes,
+ * consistent with {@link #fieldValueToString(Object)} and the streaming
extraction path.
+ */
+ private static String fieldNodeToString(final JsonNode node) {
+ if (node == null || node.isNull() || !node.isValueNode()) {
+ return null;
+ }
+ return node.asText();
+ }
+
+ /**
+ * Whether the streaming-parser token represents a usable scalar value
(text, number, boolean)
+ * rather than a container start ({@code {}/[]}) or a JSON null. Keeps the
streaming ID/index
+ * extraction consistent with the Map and JsonNode paths, which treat
objects/arrays/nulls as
+ * absent fields.
+ */
+ private static boolean isScalarValue(final JsonToken token) {
+ return token != null && !token.isStructStart() && token !=
JsonToken.VALUE_NULL;
+ }
+
/**
* Reads a processor property as a JSON Object string and deserializes it
into a Map.
* Returns an empty Map when the property is blank. Throws {@link
ProcessException} if the
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 f023920120e..b93fe826720 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
@@ -28,9 +28,11 @@ import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
@@ -881,4 +883,231 @@ public class PutElasticsearchJsonTest extends
AbstractPutElasticsearchTest {
runner.assertTransferCount(PutElasticsearchJson.REL_BULK_REQUEST, 0);
}
+
+ //
-------------------------------------------------------------------------
+ // Index Field / Timestamp Field / Retain toggles (NIFI-15985)
+ //
-------------------------------------------------------------------------
+
+ /** Registers a consumer that collects every operation sent to the _bulk
API into the returned list. */
+ private List<IndexOperationRequest> captureOperations() {
+ final List<IndexOperationRequest> captured = new ArrayList<>();
+ clientService.setEvalConsumer((final List<IndexOperationRequest>
items) -> captured.addAll(items));
+ return captured;
+ }
+
+ /** Returns the document body of an operation as a String, from raw JSON
bytes or the fields Map. */
+ private static String docContent(final IndexOperationRequest item) {
+ if (item.getRawJsonBytes() != null) {
+ return new String(item.getRawJsonBytes(), StandardCharsets.UTF_8);
+ }
+ return item.getFields() == null ? "" : item.getFields().toString();
+ }
+
+ @Test
+ void testIndexFieldNdjsonExtractionRetainedByDefault() {
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "target_index");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"target_index\":\"docs-2026\",\"msg\":\"hello\"}\n");
+ runner.run();
+
+ runner.assertTransferCount(AbstractPutElasticsearch.REL_SUCCESSFUL, 1);
+ assertEquals(1, ops.size());
+ assertEquals("docs-2026", ops.getFirst().getIndex());
+ // Retain Index Field defaults to true → the field stays in the
document body
+ assertTrue(docContent(ops.getFirst()).contains("target_index"));
+ }
+
+ @Test
+ void testIndexFieldNdjsonRemovedWhenRetainFalse() {
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "target_index");
+ runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"target_index\":\"docs-2026\",\"msg\":\"hello\"}\n");
+ runner.run();
+
+ assertEquals("docs-2026", ops.getFirst().getIndex());
+ final String content = docContent(ops.getFirst());
+ assertFalse(content.contains("target_index"), "Index Field should be
stripped when Retain Index Field=false");
+ assertTrue(content.contains("hello"), "Other fields should remain");
+ }
+
+ @Test
+ void testIndexFieldJsonArrayExtractionAndStrip() {
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.JSON_ARRAY.getValue());
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "target_index");
+ runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+
runner.enqueue("[{\"target_index\":\"docs-a\",\"msg\":\"x\"},{\"target_index\":\"docs-b\",\"msg\":\"y\"}]");
+ runner.run();
+
+ assertEquals(2, ops.size());
+ assertEquals("docs-a", ops.get(0).getIndex());
+ assertEquals("docs-b", ops.get(1).getIndex());
+ assertFalse(docContent(ops.get(0)).contains("target_index"));
+ }
+
+ @Test
+ void testIndexFieldSingleJsonExtractionRetainedByDefault() {
+ // SINGLE_JSON is the default Input Content Format
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "target_index");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"target_index\":\"docs-single\",\"msg\":\"x\"}");
+ runner.run();
+
+ assertEquals(1, ops.size());
+ assertEquals("docs-single", ops.getFirst().getIndex());
+ assertTrue(ops.getFirst().getFields().containsKey("target_index"),
"field retained by default");
+ }
+
+ @Test
+ void testIndexFieldFallsBackToIndexPropertyWhenAbsent() {
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "target_index");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"msg\":\"no index field here\"}\n");
+ runner.run();
+
+ assertEquals("test_index", ops.getFirst().getIndex(), "Should fall
back to the Index property when the field is absent");
+ }
+
+ @Test
+ void testTimestampFieldNdjsonCopiesToAtTimestampAndStrips() {
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.TIMESTAMP_FIELD, "event_time");
+ runner.setProperty(PutElasticsearchJson.RETAIN_TIMESTAMP_FIELD,
"false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+
runner.enqueue("{\"event_time\":\"2026-01-02T03:04:05Z\",\"msg\":\"x\"}\n");
+ runner.run();
+
+ final String content = docContent(ops.getFirst());
+ assertTrue(content.contains("@timestamp"), "@timestamp should be
added");
+ assertTrue(content.contains("2026-01-02T03:04:05Z"));
+ assertFalse(content.contains("event_time"), "source timestamp field
removed when retain=false");
+ }
+
+ @Test
+ void testTimestampFieldJsonArrayStrip() {
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.JSON_ARRAY.getValue());
+ runner.setProperty(PutElasticsearchJson.TIMESTAMP_FIELD, "event_time");
+ runner.setProperty(PutElasticsearchJson.RETAIN_TIMESTAMP_FIELD,
"false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+
runner.enqueue("[{\"event_time\":\"2026-05-31T00:00:00Z\",\"msg\":\"x\"}]");
+ runner.run();
+
+ final String content = docContent(ops.getFirst());
+ assertTrue(content.contains("@timestamp"));
+ assertFalse(content.contains("event_time"));
+ }
+
+ @Test
+ void testTimestampFieldSingleJsonRetainedByDefault() {
+ runner.setProperty(PutElasticsearchJson.TIMESTAMP_FIELD, "event_time");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+
runner.enqueue("{\"event_time\":\"2026-01-02T03:04:05Z\",\"msg\":\"x\"}");
+ runner.run();
+
+ final Map<String, Object> fields = ops.getFirst().getFields();
+ assertEquals("2026-01-02T03:04:05Z", fields.get("@timestamp"));
+ assertTrue(fields.containsKey("event_time"), "source field retained by
default");
+ }
+
+ @Test
+ void testIdentifierFieldRetainedByDefaultNdjson() {
+ // Backward compatibility: before this feature the Identifier Field
was always kept in the document
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.IDENTIFIER_FIELD, "id");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"id\":\"abc\",\"msg\":\"x\"}\n");
+ runner.run();
+
+ assertEquals("abc", ops.getFirst().getId());
+ assertTrue(docContent(ops.getFirst()).contains("\"id\""), "Identifier
Field kept by default for backward compatibility");
+ }
+
+ @Test
+ void testIdentifierFieldRemovedWhenRetainFalseNdjson() {
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.IDENTIFIER_FIELD, "id");
+ runner.setProperty(PutElasticsearchJson.RETAIN_IDENTIFIER_FIELD,
"false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"id\":\"abc\",\"msg\":\"x\"}\n");
+ runner.run();
+
+ assertEquals("abc", ops.getFirst().getId());
+ final String content = docContent(ops.getFirst());
+ assertFalse(content.contains("\"id\""), "identifier field removed when
retain=false");
+ assertTrue(content.contains("msg"));
+ }
+
+ @Test
+ void testIndexFieldNumericValueConsistentAcrossFormats() {
+ // The Map (Single JSON), JsonNode (JSON Array), and streaming
(NDJSON) paths must all
+ // resolve a non-string index value to the same string.
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "idx");
+ List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"idx\":12345,\"msg\":\"x\"}");
+ runner.run();
+ assertEquals("12345", ops.getFirst().getIndex(), "Single JSON Map
path");
+ runner.clearTransferState();
+
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.JSON_ARRAY.getValue());
+ ops = captureOperations();
+ runner.enqueue("[{\"idx\":12345,\"msg\":\"x\"}]");
+ runner.run();
+ assertEquals("12345", ops.getFirst().getIndex(), "JSON Array JsonNode
path");
+ runner.clearTransferState();
+
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ ops = captureOperations();
+ runner.enqueue("{\"idx\":12345,\"msg\":\"x\"}\n");
+ runner.run();
+ assertEquals("12345", ops.getFirst().getIndex(), "NDJSON streaming
path");
+ }
+
+ @Test
+ void testIndexFieldNullValueFallsBackConsistently() {
+ // A JSON null index value must fall back to the Index property on
every path —
+ // in particular the NDJSON streaming path must not emit the literal
string "null".
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "idx");
+ List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"idx\":null,\"msg\":\"x\"}");
+ runner.run();
+ assertEquals("test_index", ops.getFirst().getIndex(), "Single JSON
null -> fallback");
+ runner.clearTransferState();
+
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.JSON_ARRAY.getValue());
+ ops = captureOperations();
+ runner.enqueue("[{\"idx\":null,\"msg\":\"x\"}]");
+ runner.run();
+ assertEquals("test_index", ops.getFirst().getIndex(), "JSON Array null
-> fallback");
+ runner.clearTransferState();
+
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ ops = captureOperations();
+ runner.enqueue("{\"idx\":null,\"msg\":\"x\"}\n");
+ runner.run();
+ assertEquals("test_index", ops.getFirst().getIndex(), "NDJSON null ->
fallback (not literal \"null\")");
+ }
}
