You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by GitBox <gi...@apache.org> on 2021/10/05 11:32:55 UTC
[GitHub] [nifi] gresockj commented on a change in pull request #4691: NIFI-7990 add properties to map Record field as @timestamp in output …
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org