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 7ae29ec40ba NIFI-16028 - Support nested field paths in
PutElasticsearchJson (#11355)
7ae29ec40ba is described below
commit 7ae29ec40ba106bf79f67fb466f91b85473540fc
Author: agturley <[email protected]>
AuthorDate: Fri Jul 10 22:04:22 2026 -0700
NIFI-16028 - Support nested field paths in PutElasticsearchJson (#11355)
Allow the Index Field, Identifier Field, and Timestamp Field properties to
reference a nested field with a "/"-delimited path (e.g. "@metadata/index").
A value with no "/" behaves exactly as before (top-level lookup), so
existing
flows are unaffected.
When the corresponding Retain property is false, the extracted leaf is
removed
and any ancestor object left empty is pruned recursively, so a transient
envelope like "@metadata" is not indexed once its fields are consumed;
parents
that still have other fields are kept.
A field name that literally contains a "/" can be addressed by escaping the
slash as "\/" (and a literal backslash as "\\"); a backslash not followed by
"/" or "\" is left untouched, so existing field names are unaffected. The
escaping rule and worked nesting/escaping examples are documented in the
Index, Identifier, and Timestamp Field property descriptions.
Applies across NDJSON, JSON Array, and Single JSON, and all index
operations.
The NDJSON raw-bytes fast path is bypassed only when a nested path is
configured, since the streaming scan only matches flat field names; the path
classification and decoded flat names are resolved once per onTrigger rather
than per record.
Add a Field Path Mode property to PutElasticsearchJson controlling how the
Identifier Field, Index Field, and Timestamp Field values are interpreted:
- Literal Field Name (default): the value is the exact name of a top-level
field, with "/" and "\" treated literally.
- Nested Field Path: the value is a "/"-delimited path into nested
objects,
with "\/" and "\\" escaping.
Defaulting to Literal Field Name keeps existing configurations behaving as
before, so no property migration is required. In literal mode the configured
values are escaped into single-segment paths so the shared path helpers
resolve them to the exact field name, leaving the extraction logic
unchanged.
Document both modes in additionalDetails.md and add literal-mode tests
alongside the existing nested-path coverage
---
.../elasticsearch/FieldReferenceMode.java | 57 +++
.../elasticsearch/PutElasticsearchJson.java | 377 +++++++++++++++---
.../additionalDetails.md | 56 +++
.../elasticsearch/PutElasticsearchJsonTest.java | 426 +++++++++++++++++++++
4 files changed, 872 insertions(+), 44 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/FieldReferenceMode.java
b/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/FieldReferenceMode.java
new file mode 100644
index 00000000000..e72841a96d9
--- /dev/null
+++
b/nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/FieldReferenceMode.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.elasticsearch;
+
+import org.apache.nifi.components.DescribedValue;
+
+/**
+ * Controls how the Identifier Field, Index Field, and Timestamp Field
property values are
+ * interpreted when locating a field within a document.
+ */
+public enum FieldReferenceMode implements DescribedValue {
+ LITERAL("Literal Field Name", "Literal Field Name",
+ "The configured value is the exact name of a top-level field in
the document; \"/\" and \"\\\" "
+ + "are treated as literal characters. This is the default
and matches the behavior of earlier releases."),
+ NESTED_PATH("Nested Field Path", "Nested Field Path",
+ "The configured value is a \"/\"-delimited path into nested
objects (e.g. \"@metadata/id\"). "
+ + "To reference a field name that contains a literal
\"/\", escape it as \"\\/\"; a literal backslash is \"\\\\\".");
+
+ private final String value;
+ private final String displayName;
+ private final String description;
+
+ FieldReferenceMode(final String value, final String displayName, final
String description) {
+ this.value = value;
+ this.displayName = displayName;
+ this.description = description;
+ }
+
+ @Override
+ public String getValue() {
+ return value;
+ }
+
+ @Override
+ public String getDisplayName() {
+ return displayName;
+ }
+
+ @Override
+ public String getDescription() {
+ return description;
+ }
+}
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 29b05d854c1..1ee5c936b09 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
@@ -237,10 +237,25 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
.dependsOn(INPUT_FORMAT, InputFormat.SINGLE_JSON)
.build();
+ static final PropertyDescriptor FIELD_PATH_MODE = new
PropertyDescriptor.Builder()
+ .name("Field Path Mode")
+ .description("""
+ How the Identifier Field, Index Field, and Timestamp Field
property values are interpreted \
+ when locating a field within each document. In "Literal
Field Name" mode (the default) each \
+ value is the exact name of a top-level field. In "Nested
Field Path" mode each value is a \
+ "/"-delimited path into nested objects; see the
processor's Additional Details for the full \
+ path syntax, including how to reference a field name that
contains a literal "/".\
+ """)
+ .required(true)
+ .allowableValues(FieldReferenceMode.class)
+ .defaultValue(FieldReferenceMode.LITERAL.getValue())
+ .build();
+
static final PropertyDescriptor IDENTIFIER_FIELD = new
PropertyDescriptor.Builder()
.name("Identifier Field")
.description("""
- The name of the field within each document to use as the
Elasticsearch document ID. \
+ The name of the field within each document to use as the
Elasticsearch document ID, \
+ interpreted as a literal field name or a nested
"/"-delimited path per the Field Path Mode property. \
If the field is not present in a document or this property
is left blank, no document ID is set \
and Elasticsearch will auto-generate one.\
""")
@@ -255,7 +270,8 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
.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.\
+ 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")
@@ -266,7 +282,8 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
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. \
+ The name of the field within each document to use as the
Elasticsearch index name, \
+ interpreted as a literal field name or a nested
"/"-delimited path per the Field Path Mode property. \
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.\
""")
@@ -280,7 +297,8 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
.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.\
+ 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")
@@ -292,7 +310,8 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
.name("Timestamp Field")
.description("""
The name of a field within each document whose value will
be written to \
- Elasticsearch as the @timestamp field. \
+ Elasticsearch as the @timestamp field, interpreted as a
literal field name or a nested \
+ "/"-delimited path per the Field Path Mode property. \
If the field is absent or this property is left blank, no
@timestamp is set.\
""")
.required(false)
@@ -305,7 +324,8 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
.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.\
+ 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")
@@ -341,6 +361,7 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
MAX_BATCH_SIZE,
SUPPRESS_NULLS,
ID_ATTRIBUTE,
+ FIELD_PATH_MODE,
IDENTIFIER_FIELD,
RETAIN_IDENTIFIER_FIELD,
INDEX_FIELD,
@@ -466,11 +487,20 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
final String idAttribute = inputFormat == InputFormat.SINGLE_JSON
? context.getProperty(ID_ATTRIBUTE).getValue()
: null;
+ // In Literal Field Name mode the configured values are escaped into
single-segment paths so the
+ // shared path helpers resolve them to the exact (verbatim) field
name; in Nested Field Path mode
+ // they are already "/"-delimited paths and used as-is.
+ final boolean nestedFieldPaths =
FieldReferenceMode.NESTED_PATH.getValue().equals(context.getProperty(FIELD_PATH_MODE).getValue());
final String documentIdField = inputFormat != InputFormat.SINGLE_JSON
- ?
context.getProperty(IDENTIFIER_FIELD).evaluateAttributeExpressions().getValue()
+ ?
fieldPath(context.getProperty(IDENTIFIER_FIELD).evaluateAttributeExpressions().getValue(),
nestedFieldPaths)
: null;
- final String documentIndexField =
context.getProperty(INDEX_FIELD).evaluateAttributeExpressions().getValue();
- final String documentTimestampField =
context.getProperty(TIMESTAMP_FIELD).evaluateAttributeExpressions().getValue();
+ final String documentIndexField =
fieldPath(context.getProperty(INDEX_FIELD).evaluateAttributeExpressions().getValue(),
nestedFieldPaths);
+ final String documentTimestampField =
fieldPath(context.getProperty(TIMESTAMP_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);
+ final String documentIdFieldName = documentIdField == null ? null :
decodeSegment(documentIdField);
+ final String documentIndexFieldName = documentIndexField == null ?
null : decodeSegment(documentIndexField);
// 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.
@@ -541,16 +571,18 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
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) {
+ // 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 Map<String, Object> contentMap =
mapReader.readValue(trimmedLine);
id = resolveId(contentMap,
documentIdField, flowFileIdAttribute);
docIndex = resolveIndex(contentMap,
documentIndexField, index);
if (stripId) {
- contentMap.remove(documentIdField);
+ removeAtPath(contentMap,
documentIdField);
}
if (stripIdx) {
- contentMap.remove(documentIndexField);
+ removeAtPath(contentMap,
documentIndexField);
}
applyTimestamp(contentMap,
documentTimestampField, retainTimestampField);
rawJsonBytes = suppressingWriter != null
@@ -558,7 +590,8 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
:
mapper.writeValueAsBytes(contentMap);
} else {
// Raw-bytes path: single streaming scan
finds both fields at once.
- final String[] extracted =
extractIdAndIndex(trimmedLine, documentIdField, flowFileIdAttribute,
documentIndexField, index);
+ // The field names are pre-decoded and
known to be flat (non-nested) here.
+ final String[] extracted =
extractIdAndIndex(trimmedLine, documentIdFieldName, flowFileIdAttribute,
documentIndexFieldName, index);
id = extracted[0];
docIndex = extracted[1];
rawJsonBytes =
trimmedLine.getBytes(StandardCharsets.UTF_8);
@@ -580,10 +613,10 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
final String id = resolveId(contentMap,
documentIdField, flowFileIdAttribute);
final String docIndex =
resolveIndex(contentMap, documentIndexField, index);
if (!retainIdentifierField &&
StringUtils.isNotBlank(documentIdField)) {
- contentMap.remove(documentIdField);
+ removeAtPath(contentMap, documentIdField);
}
if (!retainIndexField &&
StringUtils.isNotBlank(documentIndexField)) {
- contentMap.remove(documentIndexField);
+ removeAtPath(contentMap,
documentIndexField);
}
applyTimestamp(contentMap,
documentTimestampField, retainTimestampField);
opRequest = IndexOperationRequest.builder()
@@ -649,10 +682,10 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
id = resolveId(contentMap,
documentIdField, flowFileIdAttribute);
docIndex = resolveIndex(contentMap,
documentIndexField, index);
if (!retainIdentifierField &&
StringUtils.isNotBlank(documentIdField)) {
- contentMap.remove(documentIdField);
+ removeAtPath(contentMap,
documentIdField);
}
if (!retainIndexField &&
StringUtils.isNotBlank(documentIndexField)) {
-
contentMap.remove(documentIndexField);
+ removeAtPath(contentMap,
documentIndexField);
}
applyTimestamp(contentMap,
documentTimestampField, retainTimestampField);
rawJsonBytes =
suppressingWriter.writeValueAsBytes(contentMap);
@@ -667,10 +700,10 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
if (node.isObject()) {
final ObjectNode objectNode =
(ObjectNode) node;
if (!retainIdentifierField &&
StringUtils.isNotBlank(documentIdField)) {
-
objectNode.remove(documentIdField);
+ removeAtPath(objectNode,
documentIdField);
}
if (!retainIndexField &&
StringUtils.isNotBlank(documentIndexField)) {
-
objectNode.remove(documentIndexField);
+ removeAtPath(objectNode,
documentIndexField);
}
applyTimestamp(objectNode,
documentTimestampField, retainTimestampField);
}
@@ -695,10 +728,10 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
final String id = resolveId(contentMap,
documentIdField, flowFileIdAttribute);
final String docIndex =
resolveIndex(contentMap, documentIndexField, index);
if (!retainIdentifierField &&
StringUtils.isNotBlank(documentIdField)) {
- contentMap.remove(documentIdField);
+ removeAtPath(contentMap,
documentIdField);
}
if (!retainIndexField &&
StringUtils.isNotBlank(documentIndexField)) {
- contentMap.remove(documentIndexField);
+ removeAtPath(contentMap,
documentIndexField);
}
applyTimestamp(contentMap,
documentTimestampField, retainTimestampField);
opRequest = IndexOperationRequest.builder()
@@ -736,7 +769,7 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
final String id =
StringUtils.isNotBlank(flowFileIdAttribute) ? flowFileIdAttribute : null;
final String docIndex = resolveIndex(contentMap,
documentIndexField, index);
if (!retainIndexField &&
StringUtils.isNotBlank(documentIndexField)) {
- contentMap.remove(documentIndexField);
+ removeAtPath(contentMap, documentIndexField);
}
applyTimestamp(contentMap, documentTimestampField,
retainTimestampField);
final IndexOperationRequest opRequest =
IndexOperationRequest.builder()
@@ -1029,36 +1062,38 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
}
/**
- * 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.
+ * Copies the value at {@code timestampField} to a top-level {@code
@timestamp} in the Map.
+ * The field may be a {@code /}-delimited path into nested objects (e.g.
{@code @metadata/event_time}).
+ * If {@code retain} is false, the source field is removed after copying,
pruning any ancestor
+ * objects that become empty. Does nothing when {@code timestampField} is
blank or absent.
*/
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);
+ final Object value = valueAtPath(contentMap, timestampField);
if (value != null) {
if (!retain) {
- contentMap.remove(timestampField);
+ removeAtPath(contentMap, 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.
+ * Copies the value at {@code timestampField} to a top-level {@code
@timestamp} in the ObjectNode.
+ * The field may be a {@code /}-delimited path into nested objects (e.g.
{@code @metadata/event_time}).
+ * If {@code retain} is false, the source field is removed after copying,
pruning any ancestor
+ * objects that become empty. Does nothing when {@code timestampField} is
blank or absent.
*/
private void applyTimestamp(final ObjectNode node, final String
timestampField, final boolean retain) {
if (StringUtils.isBlank(timestampField)) {
return;
}
- final JsonNode value = node.get(timestampField);
+ final JsonNode value = nodeAtPath(node, timestampField);
if (value != null && !value.isNull()) {
if (!retain) {
- node.remove(timestampField);
+ removeAtPath(node, timestampField);
}
node.set("@timestamp", value);
}
@@ -1069,13 +1104,16 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
* 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.
+ * <p>
+ * {@code idFieldName} and {@code indexFieldName} are the already-decoded
flat field names (any
+ * escaped slash/backslash resolved by the caller), compared directly
against the raw JSON keys.
*/
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);
+ final String idFieldName, final String flowFileIdAttribute,
+ final String indexFieldName, final String fallbackIndex) throws
IOException {
+ final boolean needId = StringUtils.isNotBlank(idFieldName);
+ final boolean needIndex = StringUtils.isNotBlank(indexFieldName);
String id = needId ? null :
(StringUtils.isNotBlank(flowFileIdAttribute) ? flowFileIdAttribute : null);
String docIndex = fallbackIndex;
@@ -1092,11 +1130,11 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
if (foundId && foundIndex) {
break;
}
- if (!foundId && idField.equals(p.currentName()) &&
p.nextToken() != null && isScalarValue(p.currentToken())) {
+ if (!foundId && idFieldName.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())) {
+ } else if (!foundIndex &&
indexFieldName.equals(p.currentName()) && p.nextToken() != null &&
isScalarValue(p.currentToken())) {
final String value = p.getText();
docIndex = StringUtils.isNotBlank(value) ? value :
fallbackIndex;
foundIndex = true;
@@ -1113,12 +1151,13 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
/**
* Extracts the document ID from a pre-parsed 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.
*/
private String extractId(final JsonNode node, final String idAttribute,
final String flowFileIdAttribute) {
if (StringUtils.isBlank(idAttribute)) {
return StringUtils.isNotBlank(flowFileIdAttribute) ?
flowFileIdAttribute : null;
}
- final String value = fieldNodeToString(node.get(idAttribute));
+ final String value = fieldNodeToString(nodeAtPath(node, idAttribute));
if (StringUtils.isNotBlank(value)) {
return value;
}
@@ -1127,13 +1166,14 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
/**
* Resolves the document ID for Update/Delete/Upsert operations from the
already-parsed content Map.
+ * The field may be a {@code /}-delimited path into nested objects.
* Falls back to the FlowFile attribute value when the field is absent
from the document.
*/
private String resolveId(final Map<String, Object> contentMap, final
String idAttribute, final String flowFileIdAttribute) {
if (StringUtils.isBlank(idAttribute)) {
return null;
}
- final String value = fieldValueToString(contentMap.get(idAttribute));
+ final String value = fieldValueToString(valueAtPath(contentMap,
idAttribute));
if (StringUtils.isNotBlank(value)) {
return value;
}
@@ -1143,30 +1183,279 @@ 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.
+ * The field may be a {@code /}-delimited path into nested objects.
* 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));
+ final String value = fieldNodeToString(nodeAtPath(node, 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.
+ * 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.
*/
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));
+ final String value = fieldValueToString(valueAtPath(contentMap,
indexField));
return StringUtils.isNotBlank(value) ? value : fallbackIndex;
}
+ /**
+ * Normalizes a configured Identifier/Index/Timestamp Field value into the
path form consumed by the
+ * shared path helpers. In Nested Field Path mode the value is already a
{@code /}-delimited path and
+ * is returned unchanged. In Literal Field Name mode the value names a
single field verbatim, so any
+ * {@code /} or {@code \} is escaped ({@code \} to {@code \\} then {@code
/} to {@code \/}) to produce
+ * a single-segment path that resolves back to that exact field name.
+ */
+ private static String fieldPath(final String configuredValue, final
boolean nestedFieldPaths) {
+ if (configuredValue == null || nestedFieldPaths) {
+ return configuredValue;
+ }
+ return configuredValue.replace("\\", "\\\\").replace("/", "\\/");
+ }
+
+ /**
+ * Whether {@code field} is a {@code /}-delimited nested path rather than
a flat field name.
+ * A {@code /} preceded by a backslash is an escaped literal slash within
a single field name
+ * (see {@link #splitPath(String)}) and does not make the field a nested
path.
+ */
+ private static boolean isNestedPath(final String field) {
+ if (field == null) {
+ return false;
+ }
+ for (int i = 0; i < field.length(); i++) {
+ final char c = field.charAt(i);
+ if (c == '\\' && i + 1 < field.length()) {
+ final char next = field.charAt(i + 1);
+ if (next == '/' || next == '\\') {
+ i++; // skip the escaped character
+ continue;
+ }
+ } else if (c == '/') {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Splits a {@code /}-delimited field path into its decoded segments,
ignoring empty segments
+ * produced by leading, trailing, or repeated slashes. A field name
containing no unescaped
+ * slash yields a single segment equal to the (decoded) name itself.
+ * <p>
+ * A backslash escapes the following character so it is taken literally:
{@code \/} is a slash
+ * that is part of a field name rather than a separator, and {@code \\} is
a literal backslash.
+ * A backslash before any other character (or at the end of the string) is
kept as a literal
+ * backslash, so field names containing stray backslashes are unaffected.
+ */
+ private static String[] splitPath(final String path) {
+ final List<String> segments = new ArrayList<>();
+ final StringBuilder current = new StringBuilder(path.length());
+ for (int i = 0; i < path.length(); i++) {
+ final char c = path.charAt(i);
+ if (c == '\\' && i + 1 < path.length()) {
+ final char next = path.charAt(i + 1);
+ if (next == '/' || next == '\\') {
+ current.append(next);
+ i++; // consume the escaped character
+ continue;
+ }
+ current.append(c); // lone backslash, kept literal
+ } else if (c == '/') {
+ if (current.length() > 0) {
+ segments.add(current.toString());
+ current.setLength(0);
+ }
+ } else {
+ current.append(c);
+ }
+ }
+ if (current.length() > 0) {
+ segments.add(current.toString());
+ }
+ return segments.toArray(new String[0]);
+ }
+
+ /**
+ * Decodes a single flat field name, resolving the same backslash escapes
as
+ * {@link #splitPath(String)} ({@code \/} to {@code /}, {@code \\} to
{@code \}) without treating
+ * any slash as a separator. Returns the input unchanged when it contains
no backslash.
+ */
+ private static String decodeSegment(final String field) {
+ if (field.indexOf('\\') < 0) {
+ return field;
+ }
+ final StringBuilder sb = new StringBuilder(field.length());
+ for (int i = 0; i < field.length(); i++) {
+ final char c = field.charAt(i);
+ if (c == '\\' && i + 1 < field.length()) {
+ final char next = field.charAt(i + 1);
+ if (next == '/' || next == '\\') {
+ sb.append(next);
+ i++;
+ continue;
+ }
+ }
+ sb.append(c);
+ }
+ return sb.toString();
+ }
+
+ /**
+ * Returns the value at the {@code /}-delimited {@code path} within {@code
map}, or {@code null}
+ * when any segment is missing or an intermediate segment is not a JSON
object. A path with no
+ * slash is a direct top-level lookup (no allocation).
+ */
+ private static Object valueAtPath(final Map<String, Object> map, final
String path) {
+ if (!isNestedPath(path)) {
+ return map.get(decodeSegment(path));
+ }
+ Object current = map;
+ for (final String segment : splitPath(path)) {
+ if (!(current instanceof Map)) {
+ return null;
+ }
+ current = ((Map<?, ?>) current).get(segment);
+ if (current == null) {
+ return null;
+ }
+ }
+ return current;
+ }
+
+ /**
+ * Returns the {@link JsonNode} at the {@code /}-delimited {@code path}
within {@code node}, or
+ * {@code null} when any segment is missing or an intermediate segment is
not a JSON object.
+ * A path with no slash is a direct field lookup.
+ */
+ private static JsonNode nodeAtPath(final JsonNode node, final String path)
{
+ if (!isNestedPath(path)) {
+ return node.get(decodeSegment(path));
+ }
+ JsonNode current = node;
+ for (final String segment : splitPath(path)) {
+ if (current == null || !current.isObject()) {
+ return null;
+ }
+ current = current.get(segment);
+ }
+ return current;
+ }
+
+ /**
+ * Removes the leaf at the {@code /}-delimited {@code path} within {@code
root}, then prunes any
+ * ancestor objects that become empty as a result (the transient envelope
often used for routing
+ * metadata, e.g. {@code @metadata}, should not be indexed once its fields
are extracted).
+ * Does nothing when the path is absent. A path with no slash is a direct
top-level removal.
+ */
+ private static void removeAtPath(final Map<String, Object> root, final
String path) {
+ if (!isNestedPath(path)) {
+ root.remove(decodeSegment(path));
+ return;
+ }
+ removeAtPath(root, splitPath(path), MAP_ACCESSOR);
+ }
+
+ /**
+ * Removes the leaf at the {@code /}-delimited {@code path} within {@code
root}, then prunes any
+ * ancestor objects that become empty as a result. Does nothing when the
path is absent.
+ * A path with no slash is a direct field removal.
+ */
+ private static void removeAtPath(final ObjectNode root, final String path)
{
+ if (!isNestedPath(path)) {
+ root.remove(decodeSegment(path));
+ return;
+ }
+ removeAtPath(root, splitPath(path), OBJECT_NODE_ACCESSOR);
+ }
+
+ /**
+ * Shared descend-and-prune walk for both the {@link Map} and {@link
ObjectNode} object models.
+ * Descends the parent {@code segments}, removes the leaf, then walks back
up removing each
+ * ancestor that the removal left empty. The {@link NodeAccessor} isolates
the few model-specific
+ * operations so the traversal logic exists in exactly one place.
+ */
+ private static <T> void removeAtPath(final T root, final String[]
segments, final NodeAccessor<T> accessor) {
+ final List<T> chain = new ArrayList<>();
+ T current = root;
+ chain.add(current);
+ for (int i = 0; i < segments.length - 1; i++) {
+ final T next = accessor.childObject(current, segments[i]);
+ if (next == null) {
+ return;
+ }
+ current = next;
+ chain.add(current);
+ }
+ accessor.remove(chain.get(chain.size() - 1), segments[segments.length
- 1]);
+ for (int i = chain.size() - 1; i >= 1; i--) {
+ if (accessor.isEmpty(chain.get(i))) {
+ accessor.remove(chain.get(i - 1), segments[i - 1]);
+ } else {
+ break;
+ }
+ }
+ }
+
+ /**
+ * The model-specific operations the shared {@link #removeAtPath(Object,
String[], NodeAccessor)}
+ * walk needs: fetching a child only when it is itself an object, removing
a key, and testing
+ * emptiness.
+ */
+ private interface NodeAccessor<T> {
+ /** The child object at {@code key}, or {@code null} when absent or
not itself an object. */
+ T childObject(T node, String key);
+
+ void remove(T node, String key);
+
+ boolean isEmpty(T node);
+ }
+
+ private static final NodeAccessor<Map<String, Object>> MAP_ACCESSOR = new
NodeAccessor<>() {
+ @Override
+ @SuppressWarnings("unchecked")
+ public Map<String, Object> childObject(final Map<String, Object> node,
final String key) {
+ final Object next = node.get(key);
+ return next instanceof Map ? (Map<String, Object>) next : null;
+ }
+
+ @Override
+ public void remove(final Map<String, Object> node, final String key) {
+ node.remove(key);
+ }
+
+ @Override
+ public boolean isEmpty(final Map<String, Object> node) {
+ return node.isEmpty();
+ }
+ };
+
+ private static final NodeAccessor<ObjectNode> OBJECT_NODE_ACCESSOR = new
NodeAccessor<>() {
+ @Override
+ public ObjectNode childObject(final ObjectNode node, final String key)
{
+ final JsonNode next = node.get(key);
+ return next != null && next.isObject() ? (ObjectNode) next : null;
+ }
+
+ @Override
+ public void remove(final ObjectNode node, final String key) {
+ node.remove(key);
+ }
+
+ @Override
+ public boolean isEmpty(final ObjectNode node) {
+ return node.isEmpty();
+ }
+ };
+
/**
* 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
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 a80d3c4290f..d2da2f79302 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
@@ -30,6 +30,62 @@ failed to an errors queue so that only failed FlowFiles can
be processed downstr
The index, operation and (optional) type fields are configured with default
values. The ID (optional unless the
operation is "index") can be set as an attribute on the FlowFile(s).
+### Field Path Mode
+
+The **Identifier Field**, **Index Field** and **Timestamp Field** properties
each name a field within the document
+whose value is extracted (and used as the document ID, index name, or
`@timestamp` respectively). The **Field Path
+Mode** property controls how those values are interpreted:
+
+* **Literal Field Name** (the default) — each value is the exact name of
a top-level field. A `/` or `\` is just
+ a character in the field name, so a value like `@metadata/id` matches a
top-level field literally named
+ `@metadata/id`.
+* **Nested Field Path** — each value is a `/`-delimited path into nested
objects, as described below.
+
+#### Literal Field Name (default)
+
+For the document:
+
+```json
+{
+ "@metadata/id": "abc",
+ "message": "Hello, world"
+}
+```
+
+an Identifier Field of `@metadata/id` selects `abc` — the top-level
field whose name is literally
+`@metadata/id`. The same value against `{"@metadata": {"id": "abc"}}` selects
nothing, because there is no
+top-level field with that exact name.
+
+#### Nested Field Path
+
+The remainder of this section applies when **Field Path Mode** is set to
**Nested Field Path**. For the document:
+
+```json
+{
+ "@metadata": {
+ "id": "abc",
+ "index": "my-index"
+ },
+ "message": "Hello, world"
+}
+```
+
+* the Identifier Field path `@metadata/id` selects `abc`
+* the Index Field path `@metadata/index` selects `my-index`
+
+#### Field names that contain a slash
+
+Because `/` separates path segments, a field whose name literally contains a
`/` must have that slash escaped as `\/`,
+and a literal backslash must be escaped as `\\`. For example, for the document
`{"a/b": "abc"}` the path `a\/b` selects
+the top-level field named `a/b`.
+
+#### Removal and pruning
+
+When the corresponding **Retain** property is set to `false`, the field is
removed from the document body after its
+value is extracted. For a nested path, any parent object that is left empty by
the removal is also pruned. For example,
+extracting and removing `@metadata/id` from `{"@metadata": {"id": "abc"},
"message": "Hello, world"}` leaves
+`{"message": "Hello, world"}`, with the now-empty `@metadata` object removed.
+
### 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 b93fe826720..c61efe132fb 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
@@ -43,6 +43,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -903,6 +904,47 @@ public class PutElasticsearchJsonTest extends
AbstractPutElasticsearchTest {
return item.getFields() == null ? "" : item.getFields().toString();
}
+ /** Switches the processor to Nested Field Path mode (the field properties
default to Literal Field Name). */
+ private void useNestedFieldPaths() {
+ runner.setProperty(PutElasticsearchJson.FIELD_PATH_MODE,
FieldReferenceMode.NESTED_PATH.getValue());
+ }
+
+ @Test
+ void testDocumentFieldNamesWithSlashAndBackslashPassThroughRaw() {
+ // Only the configured Identifier/Index/Timestamp Field property
values are parsed as nested
+ // paths; field names in the document body are never interpreted. With
no extraction configured,
+ // the raw bytes -- including field names that contain "/" and "\" --
are passed through unchanged.
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+ final String doc =
"{\"a/b\":\"x\",\"c\\\\d\":\"y\",\"msg\":\"hello\"}";
+ runner.enqueue(doc + "\n");
+ runner.run();
+
+ runner.assertTransferCount(AbstractPutElasticsearch.REL_SUCCESSFUL, 1);
+ assertEquals(doc, docContent(ops.getFirst()), "document body passed
through byte-for-byte");
+ }
+
+ @Test
+ void testUnrelatedSlashFieldRetainedWhenExtractingId() {
+ // Even on the parse-and-re-serialize path (here forced by removing
the extracted id field), a
+ // document field whose name contains a "/" is not the configured path
and is left untouched.
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.IDENTIFIER_FIELD, "doc_id");
+ runner.setProperty(PutElasticsearchJson.RETAIN_IDENTIFIER_FIELD,
"false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"doc_id\":\"1\",\"a/b\":\"x\",\"msg\":\"hello\"}\n");
+ runner.run();
+
+ assertEquals("1", ops.getFirst().getId());
+ final String content = docContent(ops.getFirst());
+ assertTrue(content.contains("\"a/b\""), "unrelated slash-named field
retained");
+ assertFalse(content.contains("doc_id"), "only the configured field is
removed");
+ }
+
@Test
void testIndexFieldNdjsonExtractionRetainedByDefault() {
runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
@@ -1110,4 +1152,388 @@ public class PutElasticsearchJsonTest extends
AbstractPutElasticsearchTest {
runner.run();
assertEquals("test_index", ops.getFirst().getIndex(), "NDJSON null ->
fallback (not literal \"null\")");
}
+
+ //
-------------------------------------------------------------------------
+ // Field Path Mode: Literal Field Name (the default)
+ //
-------------------------------------------------------------------------
+
+ @Test
+ void testFieldPathModeDefaultsToLiteral() {
+ assertEquals(FieldReferenceMode.LITERAL.getValue(),
PutElasticsearchJson.FIELD_PATH_MODE.getDefaultValue());
+ }
+
+ @Test
+ void testLiteralModeTreatsSlashAsLiteralFieldName() {
+ // Default (Literal Field Name) mode: "@metadata/id" names one
top-level field, so a document whose
+ // id is nested under @metadata is NOT matched -- this is the
pre-nested-path behavior preserved on upgrade.
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.IDENTIFIER_FIELD,
"@metadata/id");
+ runner.setProperty(PutElasticsearchJson.RETAIN_IDENTIFIER_FIELD,
"false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"@metadata\":{\"id\":\"abc-123\"},\"msg\":\"x\"}\n");
+ runner.run();
+
+ assertNull(ops.getFirst().getId(), "literal mode does not descend into
@metadata");
+ assertTrue(docContent(ops.getFirst()).contains("@metadata"), "nested
object untouched in literal mode");
+ }
+
+ @Test
+ void testLiteralModeMatchesFieldNameContainingSlash() {
+ // In literal mode the value names the field verbatim, including a
"/", with no escaping required.
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "a/b");
+ runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"a/b\":\"docs-x\",\"msg\":\"m\"}\n");
+ runner.run();
+
+ assertEquals("docs-x", ops.getFirst().getIndex(), "literal field named
a/b matched verbatim");
+ assertFalse(docContent(ops.getFirst()).contains("a/b"), "literal field
removed");
+ }
+
+ //
-------------------------------------------------------------------------
+ // Nested field paths (e.g. @metadata/index)
+ //
-------------------------------------------------------------------------
+
+ @Test
+ void testNestedIndexFieldNdjsonPrunesEmptyParent() {
+ useNestedFieldPaths();
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD,
"@metadata/index");
+ runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+
runner.enqueue("{\"@metadata\":{\"index\":\"docs-2026\"},\"msg\":\"hello\"}\n");
+ runner.run();
+
+ assertEquals("docs-2026", ops.getFirst().getIndex());
+ final String content = docContent(ops.getFirst());
+ assertFalse(content.contains("@metadata"), "now-empty parent should be
pruned");
+ assertTrue(content.contains("hello"));
+ }
+
+ @Test
+ void testNestedIndexFieldJsonArrayPrunesEmptyParent() {
+ useNestedFieldPaths();
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.JSON_ARRAY.getValue());
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD,
"@metadata/index");
+ runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+
runner.enqueue("[{\"@metadata\":{\"index\":\"docs-a\"},\"msg\":\"x\"}]");
+ runner.run();
+
+ assertEquals("docs-a", ops.getFirst().getIndex());
+ assertFalse(docContent(ops.getFirst()).contains("@metadata"));
+ }
+
+ @Test
+ void testNestedIndexFieldSingleJsonPrunesEmptyParent() {
+ useNestedFieldPaths();
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD,
"@metadata/index");
+ runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+
runner.enqueue("{\"@metadata\":{\"index\":\"docs-single\"},\"msg\":\"x\"}");
+ runner.run();
+
+ assertEquals("docs-single", ops.getFirst().getIndex());
+ assertFalse(ops.getFirst().getFields().containsKey("@metadata"));
+ }
+
+ @Test
+ void testNestedIndexFieldKeepsParentWithRemainingSiblings() {
+ useNestedFieldPaths();
+ // @metadata still holds another field after the index leaf is
removed, so it must NOT be pruned
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD,
"@metadata/index");
+ runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+
runner.enqueue("{\"@metadata\":{\"index\":\"docs-2026\",\"keep\":\"yes\"},\"msg\":\"hello\"}\n");
+ runner.run();
+
+ assertEquals("docs-2026", ops.getFirst().getIndex());
+ final String content = docContent(ops.getFirst());
+ assertTrue(content.contains("@metadata"), "parent kept because it
still has siblings");
+ assertTrue(content.contains("keep"));
+ assertFalse(content.contains("\"index\""));
+ }
+
+ @Test
+ void testNestedIndexFieldRetainedByDefault() {
+ useNestedFieldPaths();
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD,
"@metadata/index");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+
runner.enqueue("{\"@metadata\":{\"index\":\"docs-2026\"},\"msg\":\"hello\"}\n");
+ runner.run();
+
+ assertEquals("docs-2026", ops.getFirst().getIndex());
+ assertTrue(docContent(ops.getFirst()).contains("@metadata"), "nested
field retained by default");
+ }
+
+ @Test
+ void testNestedIdentifierFieldNdjsonPrunesEmptyParent() {
+ useNestedFieldPaths();
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.IDENTIFIER_FIELD,
"@metadata/id");
+ runner.setProperty(PutElasticsearchJson.RETAIN_IDENTIFIER_FIELD,
"false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"@metadata\":{\"id\":\"abc-123\"},\"msg\":\"x\"}\n");
+ runner.run();
+
+ assertEquals("abc-123", ops.getFirst().getId());
+ assertFalse(docContent(ops.getFirst()).contains("@metadata"));
+ }
+
+ @Test
+ void testNestedTimestampFieldNdjsonPrunesEmptyParent() {
+ useNestedFieldPaths();
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.TIMESTAMP_FIELD,
"@metadata/event_time");
+ runner.setProperty(PutElasticsearchJson.RETAIN_TIMESTAMP_FIELD,
"false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+
runner.enqueue("{\"@metadata\":{\"event_time\":\"2026-01-02T03:04:05Z\"},\"msg\":\"x\"}\n");
+ runner.run();
+
+ final String content = docContent(ops.getFirst());
+ assertTrue(content.contains("@timestamp"));
+ assertTrue(content.contains("2026-01-02T03:04:05Z"));
+ assertFalse(content.contains("@metadata"), "empty parent pruned after
extracting nested timestamp");
+ }
+
+ @Test
+ void testNestedTimestampFieldSingleJsonPrunesEmptyParent() {
+ useNestedFieldPaths();
+ runner.setProperty(PutElasticsearchJson.TIMESTAMP_FIELD, "meta/ts");
+ runner.setProperty(PutElasticsearchJson.RETAIN_TIMESTAMP_FIELD,
"false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+
runner.enqueue("{\"meta\":{\"ts\":\"2026-05-31T00:00:00Z\"},\"msg\":\"x\"}");
+ runner.run();
+
+ final Map<String, Object> fields = ops.getFirst().getFields();
+ assertEquals("2026-05-31T00:00:00Z", fields.get("@timestamp"));
+ assertFalse(fields.containsKey("meta"), "empty parent pruned");
+ }
+
+ @Test
+ void testNestedIndexFieldMissingLeafFallsBack() {
+ useNestedFieldPaths();
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD,
"@metadata/index");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"@metadata\":{\"other\":\"x\"},\"msg\":\"y\"}\n");
+ runner.run();
+
+ assertEquals("test_index", ops.getFirst().getIndex(), "missing nested
leaf -> fall back to Index property");
+ }
+
+ @Test
+ void testNestedIndexFieldConsistentAcrossFormats() {
+ useNestedFieldPaths();
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD,
"@metadata/index");
+ List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"@metadata\":{\"index\":\"docs-x\"},\"msg\":\"a\"}");
+ runner.run();
+ assertEquals("docs-x", ops.getFirst().getIndex(), "Single JSON nested
path");
+ runner.clearTransferState();
+
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.JSON_ARRAY.getValue());
+ ops = captureOperations();
+
runner.enqueue("[{\"@metadata\":{\"index\":\"docs-x\"},\"msg\":\"a\"}]");
+ runner.run();
+ assertEquals("docs-x", ops.getFirst().getIndex(), "JSON Array nested
path");
+ runner.clearTransferState();
+
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ ops = captureOperations();
+
runner.enqueue("{\"@metadata\":{\"index\":\"docs-x\"},\"msg\":\"a\"}\n");
+ runner.run();
+ assertEquals("docs-x", ops.getFirst().getIndex(), "NDJSON nested
path");
+ }
+
+ @Test
+ void testDeeplyNestedIndexFieldPrunesAllEmptyAncestors() {
+ useNestedFieldPaths();
+ // A 3-level path (a/b/c): removing the leaf must recursively prune b
and then a,
+ // since each ancestor becomes empty in turn.
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "a/b/c");
+ runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+
runner.enqueue("{\"a\":{\"b\":{\"c\":\"docs-deep\"}},\"msg\":\"hello\"}\n");
+ runner.run();
+
+ assertEquals("docs-deep", ops.getFirst().getIndex());
+ final String content = docContent(ops.getFirst());
+ assertFalse(content.contains("\"a\""), "all empty ancestor objects
pruned recursively");
+ assertFalse(content.contains("\"b\""));
+ assertFalse(content.contains("\"c\""));
+ assertTrue(content.contains("hello"));
+ }
+
+ @Test
+ void testDeeplyNestedIndexFieldKeepsAncestorWithRemainingSibling() {
+ useNestedFieldPaths();
+ // A 3-level path (a/b/c) where the middle object 'b' retains a
sibling after the leaf is
+ // removed: pruning must stop at 'b' and leave 'a' intact.
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "a/b/c");
+ runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+
runner.enqueue("{\"a\":{\"b\":{\"c\":\"docs-deep\",\"keep\":\"yes\"}},\"msg\":\"hello\"}\n");
+ runner.run();
+
+ assertEquals("docs-deep", ops.getFirst().getIndex());
+ final String content = docContent(ops.getFirst());
+ assertTrue(content.contains("\"keep\""), "sibling stops the recursive
pruning");
+ assertTrue(content.contains("\"b\""));
+ assertTrue(content.contains("\"a\""));
+ assertFalse(content.contains("\"c\""));
+ }
+
+ @Test
+ void testNestedIdentifierFieldUpdateOperationResolvesAndPrunes() {
+ useNestedFieldPaths();
+ // Update/Delete/Upsert resolve the ID from the parsed content Map via
resolveId, a different
+ // code path than the Index/Create streaming/JsonNode extraction.
Verify the nested path is
+ // honored and the now-empty parent is pruned from the doc body.
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.INDEX_OP,
IndexOperationRequest.Operation.Update.getValue().toLowerCase());
+ runner.setProperty(PutElasticsearchJson.IDENTIFIER_FIELD,
"@metadata/id");
+ runner.setProperty(PutElasticsearchJson.RETAIN_IDENTIFIER_FIELD,
"false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"@metadata\":{\"id\":\"abc-123\"},\"msg\":\"x\"}\n");
+ runner.run();
+
+ assertEquals("abc-123", ops.getFirst().getId());
+ assertFalse(ops.getFirst().getFields().containsKey("@metadata"),
"empty parent pruned after extracting nested id");
+ assertTrue(ops.getFirst().getFields().containsKey("msg"));
+ }
+
+ //
-------------------------------------------------------------------------
+ // Escaped path separators (e.g. a\/b addresses the flat field named "a/b")
+ //
-------------------------------------------------------------------------
+
+ @Test
+ void testEscapedSlashIndexFieldTreatedAsFlatNameAcrossFormats() {
+ useNestedFieldPaths();
+ // "a\/b" must resolve the top-level field literally named "a/b" on
every path,
+ // including the NDJSON raw-bytes streaming scan.
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "a\\/b");
+ List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"a/b\":\"docs-x\",\"msg\":\"m\"}");
+ runner.run();
+ assertEquals("docs-x", ops.getFirst().getIndex(), "Single JSON
escaped-slash flat field");
+ runner.clearTransferState();
+
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.JSON_ARRAY.getValue());
+ ops = captureOperations();
+ runner.enqueue("[{\"a/b\":\"docs-x\",\"msg\":\"m\"}]");
+ runner.run();
+ assertEquals("docs-x", ops.getFirst().getIndex(), "JSON Array
escaped-slash flat field");
+ runner.clearTransferState();
+
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ ops = captureOperations();
+ runner.enqueue("{\"a/b\":\"docs-x\",\"msg\":\"m\"}\n");
+ runner.run();
+ assertEquals("docs-x", ops.getFirst().getIndex(), "NDJSON
escaped-slash flat field (raw streaming path)");
+ }
+
+ @Test
+ void testEscapedSlashIndexFieldRemovedWhenRetainFalse() {
+ useNestedFieldPaths();
+ // The escaped-slash field is flat, so removal must delete the
top-level "a/b" key and must
+ // NOT descend into a nonexistent "a" object.
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "a\\/b");
+ runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"a/b\":\"docs-x\",\"keep\":\"y\"}\n");
+ runner.run();
+
+ assertEquals("docs-x", ops.getFirst().getIndex());
+ final String content = docContent(ops.getFirst());
+ assertFalse(content.contains("a/b"), "escaped-slash flat field
removed");
+ assertTrue(content.contains("keep"));
+ }
+
+ @Test
+ void testNestedPathWithEscapedSlashSegment() {
+ useNestedFieldPaths();
+ // "a\/b/c": the first segment is the literal field "a/b", then nested
field "c".
+ // Removing the leaf prunes the now-empty "a/b" object.
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "a\\/b/c");
+ runner.setProperty(PutElasticsearchJson.RETAIN_INDEX_FIELD, "false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"a/b\":{\"c\":\"docs-x\"},\"msg\":\"m\"}\n");
+ runner.run();
+
+ assertEquals("docs-x", ops.getFirst().getIndex());
+ final String content = docContent(ops.getFirst());
+ assertFalse(content.contains("a/b"), "now-empty object named \"a/b\"
pruned");
+ assertTrue(content.contains("msg"));
+ }
+
+ @Test
+ void testEscapedBackslashFieldName() {
+ useNestedFieldPaths();
+ // "a\\b" addresses the flat field literally named "a\b" (single
backslash).
+ runner.setProperty(PutElasticsearchJson.INDEX_FIELD, "a\\\\b");
+ final List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"a\\\\b\":\"docs-x\",\"msg\":\"m\"}");
+ runner.run();
+ assertEquals("docs-x", ops.getFirst().getIndex(), "\\\\ resolves to a
single literal backslash in the field name");
+ }
+
+ @Test
+ void testEscapedSlashIdentifierFieldUpdateOperation() {
+ useNestedFieldPaths();
+ // Update resolves the ID from the parsed Map via resolveId; verify
the escaped-slash flat
+ // field name is honored and removed.
+ runner.setProperty(PutElasticsearchJson.INPUT_FORMAT,
InputFormat.NDJSON.getValue());
+ runner.setProperty(PutElasticsearchJson.INDEX_OP,
IndexOperationRequest.Operation.Update.getValue().toLowerCase());
+ runner.setProperty(PutElasticsearchJson.IDENTIFIER_FIELD, "a\\/b");
+ runner.setProperty(PutElasticsearchJson.RETAIN_IDENTIFIER_FIELD,
"false");
+ runner.assertValid();
+
+ final List<IndexOperationRequest> ops = captureOperations();
+ runner.enqueue("{\"a/b\":\"id-1\",\"msg\":\"m\"}\n");
+ runner.run();
+
+ assertEquals("id-1", ops.getFirst().getId());
+ assertFalse(ops.getFirst().getFields().containsKey("a/b"),
"escaped-slash flat id field removed");
+ assertTrue(ops.getFirst().getFields().containsKey("msg"));
+ }
}
