This is an automated email from the ASF dual-hosted git repository.
mthomsen 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 db49a861b3 NIFI-10687 set Elasticsearch document _id to null if ID
attribute evaluated to blank String for PutElasticsearchRecord or
PutElasticsearchJson; use @timestamp default value if @timestamp record path
evaluates to blank String in PutElasticsearchRecord
db49a861b3 is described below
commit db49a861b3ab2e555ae0bf94e8237dd1d5fb462a
Author: Chris Sampson <[email protected]>
AuthorDate: Mon Oct 24 21:28:01 2022 +0100
NIFI-10687 set Elasticsearch document _id to null if ID attribute evaluated
to blank String for PutElasticsearchRecord or PutElasticsearchJson; use
@timestamp default value if @timestamp record path evaluates to blank String in
PutElasticsearchRecord
This closes #6575
Signed-off-by: Mike Thomsen <[email protected]>
---
.../elasticsearch/PutElasticsearchJson.java | 2 +-
.../elasticsearch/PutElasticsearchRecord.java | 6 +--
.../elasticsearch/PutElasticsearchJsonTest.groovy | 9 +++-
.../PutElasticsearchRecordTest.groovy | 51 ++++++++++++++++++++++
4 files changed, 63 insertions(+), 5 deletions(-)
diff --git
a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java
b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java
index 4f35994276..7b23d1a22d 100644
---
a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java
+++
b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java
@@ -167,7 +167,7 @@ public class PutElasticsearchJson extends
AbstractPutElasticsearch {
final String indexOp =
context.getProperty(INDEX_OP).evaluateAttributeExpressions(input).getValue();
final String index =
context.getProperty(INDEX).evaluateAttributeExpressions(input).getValue();
final String type =
context.getProperty(TYPE).evaluateAttributeExpressions(input).getValue();
- final String id = StringUtils.isNotBlank(idAttribute) ?
input.getAttribute(idAttribute) : null;
+ final String id = StringUtils.isNotBlank(idAttribute) &&
StringUtils.isNotBlank(input.getAttribute(idAttribute)) ?
input.getAttribute(idAttribute) : null;
final String charset =
context.getProperty(CHARSET).evaluateAttributeExpressions(input).getValue();
diff --git
a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
index 332c16720f..1b572815d3 100644
---
a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
+++
b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
@@ -333,10 +333,10 @@ public class PutElasticsearchRecord extends
AbstractPutElasticsearch {
final String atTimestampPath =
context.getProperty(AT_TIMESTAMP_RECORD_PATH).evaluateAttributeExpressions(input).getValue();
final RecordPath ioPath = indexOpPath != null ?
recordPathCache.getCompiled(indexOpPath) : null;
- final RecordPath path = idPath != null ?
recordPathCache.getCompiled(idPath) : null;
+ final RecordPath path = StringUtils.isNotBlank(idPath) ?
recordPathCache.getCompiled(idPath) : null;
final RecordPath iPath = indexPath != null ?
recordPathCache.getCompiled(indexPath) : null;
final RecordPath tPath = typePath != null ?
recordPathCache.getCompiled(typePath) : null;
- final RecordPath atPath = atTimestampPath != null ?
recordPathCache.getCompiled(atTimestampPath) : null;
+ final RecordPath atPath = StringUtils.isNotBlank(atTimestampPath) ?
recordPathCache.getCompiled(atTimestampPath) : null;
final boolean retainId =
context.getProperty(RETAIN_ID_FIELD).evaluateAttributeExpressions(input).asBoolean();
final boolean retainTimestamp =
context.getProperty(RETAIN_AT_TIMESTAMP_FIELD).evaluateAttributeExpressions(input).asBoolean();
@@ -562,7 +562,7 @@ public class PutElasticsearchRecord extends
AbstractPutElasticsearch {
fieldValue.updateValue(null);
}
- return fieldValue.getValue().toString();
+ return fieldValue.toString();
} else {
return fallback;
}
diff --git
a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/PutElasticsearchJsonTest.groovy
b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/PutElasticsearchJsonTest.groovy
index 096cb4f410..21068df443 100644
---
a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/PutElasticsearchJsonTest.groovy
+++
b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/PutElasticsearchJsonTest.groovy
@@ -161,7 +161,14 @@ class PutElasticsearchJsonTest {
clientService.evalParametersClosure = evalParametersClosure
- basicTest(0, 0, 1, [slices: "auto"])
+ def evalClosure = { List<IndexOperationRequest> items ->
+ int nullIdCount = items.findAll { it.id == null }.size()
+ assertEquals(1, nullIdCount)
+ }
+
+ clientService.evalClosure = evalClosure
+
+ basicTest(0, 0, 1, [slices: "auto", "doc_id": ""])
}
@Test
diff --git
a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecordTest.groovy
b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecordTest.groovy
index c81539ffb3..3c0426023a 100644
---
a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecordTest.groovy
+++
b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecordTest.groovy
@@ -478,6 +478,57 @@ class PutElasticsearchRecordTest {
runner.assertTransferCount(PutElasticsearchRecord.REL_RETRY, 0)
runner.assertTransferCount(PutElasticsearchRecord.REL_FAILED_RECORDS,
0)
runner.assertTransferCount(PutElasticsearchRecord.REL_SUCCESSFUL_RECORDS, 0)
+
+ runner.clearTransferState()
+
+ flowFileContents = prettyPrint(toJson([
+ [ id: "rec-1", op: "index", index: "bulk_a", type: "message",
msg: "Hello" ]
+ ]))
+
+ clientService.evalClosure = { List<IndexOperationRequest> items ->
+ def nullIdCount = items.findAll { it.id == null }.size()
+ def noTimestampCount = items.findAll {
it.fields.containsKey("@timestamp") }.size()
+ assertEquals(1, nullIdCount)
+ assertEquals(1, noTimestampCount)
+ }
+
+ runner.setProperty(PutElasticsearchRecord.ID_RECORD_PATH,
"\${id_not_exist}")
+ runner.setProperty(PutElasticsearchRecord.AT_TIMESTAMP_RECORD_PATH,
"\${not_exist}")
+ runner.enqueue(flowFileContents, [
+ "schema.name": "recordPathTest"
+ ])
+ runner.run()
+ runner.assertTransferCount(PutElasticsearchRecord.REL_SUCCESS, 1)
+ runner.assertTransferCount(PutElasticsearchRecord.REL_FAILURE, 0)
+ runner.assertTransferCount(PutElasticsearchRecord.REL_RETRY, 0)
+ runner.assertTransferCount(PutElasticsearchRecord.REL_FAILED_RECORDS,
0)
+
runner.assertTransferCount(PutElasticsearchRecord.REL_SUCCESSFUL_RECORDS, 0)
+
+ runner.clearTransferState()
+
+ flowFileContents = prettyPrint(toJson([
+ [ id: "rec-1", op: "index", index: "bulk_a", type: "message",
msg: "Hello", empty: "" ]
+ ]))
+
+ clientService.evalClosure = { List<IndexOperationRequest> items ->
+ def nullIdCount = items.findAll { it.id == null }.size()
+ def noTimestampCount = items.findAll {
it.fields.containsKey("@timestamp") }.size()
+ assertEquals(1, nullIdCount)
+ assertEquals(1, noTimestampCount)
+ }
+
+ runner.setProperty(PutElasticsearchRecord.ID_RECORD_PATH,
"\${will_be_empty}")
+ runner.setProperty(PutElasticsearchRecord.AT_TIMESTAMP_RECORD_PATH,
"\${will_be_empty}")
+ runner.enqueue(flowFileContents, [
+ "schema.name": "recordPathTest",
+ "will_be_empty": "/empty"
+ ])
+ runner.run()
+ runner.assertTransferCount(PutElasticsearchRecord.REL_SUCCESS, 1)
+ runner.assertTransferCount(PutElasticsearchRecord.REL_FAILURE, 0)
+ runner.assertTransferCount(PutElasticsearchRecord.REL_RETRY, 0)
+ runner.assertTransferCount(PutElasticsearchRecord.REL_FAILED_RECORDS,
0)
+
runner.assertTransferCount(PutElasticsearchRecord.REL_SUCCESSFUL_RECORDS, 0)
}
@Test
