gresockj commented on a change in pull request #4691:
URL: https://github.com/apache/nifi/pull/4691#discussion_r722097750
##########
File path:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java
##########
@@ -209,6 +207,25 @@
.required(true)
.build();
+ static final PropertyDescriptor AT_TIMESTAMP = new
PropertyDescriptor.Builder()
+ .name("put-es-record-at-timestamp")
+ .displayName("@timestamp Value")
+ .description("The value to use as the @timestamp field (required
for Elasticsearch Data Streams)")
+ .required(false)
+
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor AT_TIMESTAMP_RECORD_PATH = new
PropertyDescriptor.Builder()
+ .name("put-es-record-at-timestamp-path")
+ .displayName("@timestamp Record Path")
+ .description("A RecordPath pointing to a field in the record(s)
that contains the @timestamp for the document " +
+ "(required for Elasticsearch Data Streams). If left blank
the @timestamp will be determined using the main property type")
+ .required(false)
+ .addValidator(new RecordPathValidator())
+
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .build();
Review comment:
It looks like you use `@timestamp Value` as a default timestamp in the
unit test below -- should we rename it to `Default @timestamp`? Regardless, I
think we should make it clearer in the property descriptions what the fallback
logic is -- right now it looks like both are required properties for Data
Streams.
##########
File path:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java
##########
@@ -405,11 +424,17 @@ public void onTrigger(final ProcessContext context, final
ProcessSession session
this.nullSuppression = context.getProperty(SUPPRESS_NULLS).getValue();
- final String id_path =
context.getProperty(ID_RECORD_PATH).evaluateAttributeExpressions(flowFile).getValue();
- final RecordPath recordPath = StringUtils.isEmpty(id_path) ? null :
recordPathCache.getCompiled(id_path);
+ final String idPath =
context.getProperty(ID_RECORD_PATH).evaluateAttributeExpressions(flowFile).getValue();
+ final RecordPath recordPath = StringUtils.isEmpty(idPath) ? null :
recordPathCache.getCompiled(idPath);
final StringBuilder sb = new StringBuilder();
final Charset charset =
Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions(flowFile).getValue());
+ final String atTimestamp =
context.getProperty(AT_TIMESTAMP).evaluateAttributeExpressions(flowFile).getValue();
+ final String atTimestampPath =
context.getProperty(AT_TIMESTAMP_RECORD_PATH).isSet()
+ ?
context.getProperty(AT_TIMESTAMP_RECORD_PATH).evaluateAttributeExpressions(flowFile).getValue()
+ : null;
Review comment:
If I'm reading `StandardPropertyValue` correctly, this could be
simplified to the following since `evaluateAttributeExrpessions` gracefully
handles a null `rawValue`:
```suggestion
final String atTimestampPath =
context.getProperty(AT_TIMESTAMP_RECORD_PATH)
.evaluateAttributeExpressions(flowFile).getValue();
```
##########
File path:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java
##########
@@ -209,6 +207,25 @@
.required(true)
.build();
+ static final PropertyDescriptor AT_TIMESTAMP = new
PropertyDescriptor.Builder()
+ .name("put-es-record-at-timestamp")
+ .displayName("@timestamp Value")
+ .description("The value to use as the @timestamp field (required
for Elasticsearch Data Streams)")
+ .required(false)
+
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
+ .build();
+
+ static final PropertyDescriptor AT_TIMESTAMP_RECORD_PATH = new
PropertyDescriptor.Builder()
+ .name("put-es-record-at-timestamp-path")
+ .displayName("@timestamp Record Path")
+ .description("A RecordPath pointing to a field in the record(s)
that contains the @timestamp for the document " +
+ "(required for Elasticsearch Data Streams). If left blank
the @timestamp will be determined using the main property type")
Review comment:
What do you mean by "determined using the main property type"?
##########
File path:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestPutElasticsearchHttpRecord.java
##########
@@ -498,6 +501,132 @@ public void
testPutElasticSearchOnTriggerWithInvalidIndexOp() throws IOException
assertNotNull(out);
}
+ @Test
+ public void testPutElasticsearchOnTriggerWithNoAtTimestampPath() throws
Exception {
+ PutElasticsearchHttpRecordTestProcessor processor = new
PutElasticsearchHttpRecordTestProcessor(false);
+ runner = TestRunners.newTestRunner(processor);
+ generateTestData(1);
+ runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL,
"http://127.0.0.1:9200";);
+ runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
+
+ runner.removeProperty(PutElasticsearchHttpRecord.AT_TIMESTAMP); // no
default
+
runner.setProperty(PutElasticsearchHttpRecord.AT_TIMESTAMP_RECORD_PATH,
"/none"); // Field does not exist
+ processor.setRecordChecks(record -> assertTimestamp(record, null)); //
no @timestamp
+ runner.enqueue(new byte[0]);
+ runner.run(1, true, true);
+
+
runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
+ final MockFlowFile out =
runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
+ assertNotNull(out);
+ runner.clearTransferState();
+
+ // now add a default @timestamp
+ final String timestamp = "2020-11-27T14:37:00.000Z";
+ runner.setProperty(PutElasticsearchHttpRecord.AT_TIMESTAMP, timestamp);
+ processor.setRecordChecks(record -> assertTimestamp(record,
timestamp)); // @timestamp defaulted
+ runner.enqueue(new byte[0]);
+ runner.run(1, true, true);
+
+
runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
+ final MockFlowFile out2 =
runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
+ assertNotNull(out2);
+ }
+
+ @Test
+ public void testPutElasticsearchOnTriggerWithAtTimestampFromAttribute()
throws IOException {
+ PutElasticsearchHttpRecordTestProcessor processor = new
PutElasticsearchHttpRecordTestProcessor(false);
+ runner = TestRunners.newTestRunner(processor);
+ generateTestData(1);
+ runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL,
"http://127.0.0.1:9200";);
+ runner.setProperty(PutElasticsearchHttpRecord.INDEX, "${i}");
+ runner.setProperty(PutElasticsearchHttpRecord.AT_TIMESTAMP,
"${timestamp}");
+
+ final String timestamp = "2020-11-27T15:10:00.000Z";
+ processor.setRecordChecks(record -> assertTimestamp(record,
timestamp));
+ runner.enqueue(new byte[0], new HashMap<String, String>() {{
+ put("doc_id", "28039652144");
+ put("i", "doc");
+ put("timestamp", timestamp);
+ }});
+ runner.run(1, true, true);
+
+
runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
+ final MockFlowFile out =
runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
+ assertNotNull(out);
+ runner.clearTransferState();
+
+ // Now try an empty attribute value, should be no timestamp
+ processor.setRecordChecks(record -> assertTimestamp(record, null));
+ runner.enqueue(new byte[0], new HashMap<String, String>() {{
+ put("doc_id", "28039652144");
+ put("i", "doc");
+ }});
+ runner.run(1, true, true);
+
+
runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
+ final MockFlowFile out2 =
runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
+ assertNotNull(out2);
+ }
+
+ @Test
+ public void testPutElasticsearchOnTriggerWithAtTimstampPath() throws
Exception {
+ PutElasticsearchHttpRecordTestProcessor processor = new
PutElasticsearchHttpRecordTestProcessor(false);
+ DateTimeFormatter timeFormatter =
DateTimeFormatter.ofPattern(RecordFieldType.TIME.getDefaultFormat());
+ DateTimeFormatter dateTimeFormatter =
DateTimeFormatter.ofPattern(RecordFieldType.TIMESTAMP.getDefaultFormat());
+ DateTimeFormatter dateFormatter =
DateTimeFormatter.ofPattern(RecordFieldType.DATE.getDefaultFormat());
+ runner = TestRunners.newTestRunner(processor);
+ generateTestData(1);
+ runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL,
"http://127.0.0.1:9200";);
+ runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
+
+
runner.setProperty(PutElasticsearchHttpRecord.AT_TIMESTAMP_RECORD_PATH, "/ts");
// TIMESTAMP
+ processor.setRecordChecks(record -> assertTimestamp(record,
LOCAL_DATE_TIME.format(dateTimeFormatter)));
+ runner.enqueue(new byte[0]);
+ runner.run(1, true, true);
+
+
runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
+
assertNotNull(runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0));
+ runner.clearTransferState();
+
+
runner.setProperty(PutElasticsearchHttpRecord.AT_TIMESTAMP_RECORD_PATH,
"/date"); // DATE;
+ processor.setRecordChecks(record -> assertTimestamp(record,
LOCAL_DATE.format(dateFormatter)));
+ runner.enqueue(new byte[0]);
+ runner.run(1, true, true);
+
+
runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
+
assertNotNull(runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0));
+ runner.clearTransferState();
+
+
runner.setProperty(PutElasticsearchHttpRecord.AT_TIMESTAMP_RECORD_PATH,
"/time"); // TIME
+ processor.setRecordChecks(record -> assertTimestamp(record,
LOCAL_TIME.format(timeFormatter)));
+ runner.enqueue(new byte[0]);
+ runner.run(1, true, true);
+
+
runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
+
assertNotNull(runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0));
+ runner.clearTransferState();
+
+ // these INT/STRING values might not make sense from an Elasticsearch
point of view,
+ // but we want to prove we can handle them being selected from teh
Record
Review comment:
the*
##########
File path:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
##########
@@ -377,11 +494,92 @@ private String getFromRecordPath(final Record record,
final RecordPath path, fin
);
}
- fieldValue.updateValue(null);
+ if (!retain) {
+ fieldValue.updateValue(null);
+ }
return fieldValue.getValue().toString();
} else {
return fallback;
}
}
+
+ private Object getTimestampFromRecordPath(final Record record, final
RecordPath path, final String fallback,
+ final boolean retain) {
+ if (path == null) {
+ return coerceStringToLong("@timestamp", fallback);
+ }
+
+ final RecordPathResult result = path.evaluate(record);
+ final Optional<FieldValue> value =
result.getSelectedFields().findFirst();
+ if (value.isPresent() && value.get().getValue() != null) {
+ final FieldValue fieldValue = value.get();
+
+ final DataType dataType = fieldValue.getField().getDataType();
+ final String fieldName = fieldValue.getField().getFieldName();
+ final DataType chosenDataType = dataType.getFieldType() ==
RecordFieldType.CHOICE
+ ? DataTypeUtils.chooseDataType(value, (ChoiceDataType)
dataType)
+ : dataType;
+ final Object coercedValue =
DataTypeUtils.convertType(fieldValue.getValue(), chosenDataType, fieldName);
+ if (coercedValue == null) {
+ return null;
+ }
+
+ final Object returnValue;
+ switch (chosenDataType.getFieldType()) {
+ case DATE:
+ case TIME:
+ case TIMESTAMP:
+ final String format;
+ switch (chosenDataType.getFieldType()) {
Review comment:
How about moving the nested `switch` to a separate method for readabilty?
##########
File path:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
##########
@@ -147,9 +194,51 @@
.required(false)
.build();
+ static final PropertyDescriptor AT_TIMESTAMP_DATE_FORMAT = new
PropertyDescriptor.Builder()
+ .name("put-es-record-at-timestamp-date-format")
+ .displayName("@Timestamp Record Path Date Format")
+ .description("Specifies the format to use when writing Date field for
@timestamp. "
+ + "If not specified, the default format '" +
RecordFieldType.DATE.getDefaultFormat() + "' is used. "
+ + "If specified, the value must match the Java Simple Date
Format (for example, MM/dd/yyyy for a two-digit month, followed by "
+ + "a two-digit day, followed by a four-digit year, all
separated by '/' characters, as in 01/01/2017).")
Review comment:
Just to make the example immediately clear, let's use a date like
`01/25/2017`
##########
File path:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java
##########
@@ -266,9 +283,11 @@
descriptors.add(RECORD_WRITER);
descriptors.add(LOG_ALL_ERRORS);
descriptors.add(ID_RECORD_PATH);
+ descriptors.add(AT_TIMESTAMP_RECORD_PATH);
descriptors.add(INDEX);
descriptors.add(TYPE);
descriptors.add(INDEX_OP);
+ descriptors.add(AT_TIMESTAMP);
Review comment:
Could we put `AT_TIMESTAMP` right below `AT_TIMESTAMP_RECORD_PATH`?
##########
File path:
nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
##########
@@ -236,11 +342,18 @@ public void onTrigger(ProcessContext context,
ProcessSession session) throws Pro
final String typePath = context.getProperty(TYPE_RECORD_PATH).isSet()
?
context.getProperty(TYPE_RECORD_PATH).evaluateAttributeExpressions(input).getValue()
: null;
+ final String atTimestampPath =
context.getProperty(AT_TIMESTAMP_RECORD_PATH).isSet()
Review comment:
Same comment as above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
