Repository: nifi Updated Branches: refs/heads/master 456c9c8fc -> 6b34d3bea
NIFI-4743 - Added configurable null suppression to PutElasticsearchHttpRecord Signed-off-by: Pierre Villard <[email protected]> This closes #2501. Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/6b34d3be Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/6b34d3be Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/6b34d3be Branch: refs/heads/master Commit: 6b34d3bea9a1fd0923024018d73c3c0b0a807a67 Parents: 456c9c8 Author: Mike Thomsen <[email protected]> Authored: Thu Mar 1 08:00:03 2018 -0500 Committer: Pierre Villard <[email protected]> Committed: Mon Mar 26 20:41:21 2018 +0200 ---------------------------------------------------------------------- .../AbstractElasticsearchHttpProcessor.java | 4 +- .../PutElasticsearchHttpRecord.java | 34 ++- .../PutElasticsearchHttpRecordIT.java | 211 +++++++++++++++++++ 3 files changed, 245 insertions(+), 4 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/6b34d3be/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractElasticsearchHttpProcessor.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractElasticsearchHttpProcessor.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractElasticsearchHttpProcessor.java index 49ab682..faaf523 100644 --- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractElasticsearchHttpProcessor.java +++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractElasticsearchHttpProcessor.java @@ -121,8 +121,8 @@ public abstract class AbstractElasticsearchHttpProcessor extends AbstractElastic OkHttpClient.Builder okHttpClient = new OkHttpClient.Builder(); // Add a proxy if set - final String proxyHost = context.getProperty(PROXY_HOST).getValue(); - final Integer proxyPort = context.getProperty(PROXY_PORT).asInteger(); + final String proxyHost = context.getProperty(PROXY_HOST).evaluateAttributeExpressions().getValue(); + final Integer proxyPort = context.getProperty(PROXY_PORT).evaluateAttributeExpressions().asInteger(); if (proxyHost != null && proxyPort != null) { final Proxy proxy = new Proxy(Proxy.Type.HTTP, new InetSocketAddress(proxyHost, proxyPort)); okHttpClient.proxy(proxy); http://git-wip-us.apache.org/repos/asf/nifi/blob/6b34d3be/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java index f65abb0..ae98d52 100644 --- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java +++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java @@ -32,6 +32,7 @@ import org.apache.nifi.annotation.behavior.InputRequirement; import org.apache.nifi.annotation.documentation.CapabilityDescription; import org.apache.nifi.annotation.documentation.Tags; import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.components.AllowableValue; import org.apache.nifi.components.PropertyDescriptor; import org.apache.nifi.components.ValidationContext; import org.apache.nifi.components.ValidationResult; @@ -153,6 +154,24 @@ public class PutElasticsearchHttpRecord extends AbstractElasticsearchHttpProcess .defaultValue("index") .build(); + static final AllowableValue ALWAYS_SUPPRESS = new AllowableValue("always-suppress", "Always Suppress", + "Fields that are missing (present in the schema but not in the record), or that have a value of null, will not be written out"); + + static final AllowableValue NEVER_SUPPRESS = new AllowableValue("never-suppress", "Never Suppress", + "Fields that are missing (present in the schema but not in the record), or that have a value of null, will be written out as a null value"); + + static final AllowableValue SUPPRESS_MISSING = new AllowableValue("suppress-missing", "Suppress Missing Values", + "When a field has a value of null, it will be written out. However, if a field is defined in the schema and not present in the record, the field will not be written out."); + + static final PropertyDescriptor SUPPRESS_NULLS = new PropertyDescriptor.Builder() + .name("suppress-nulls") + .displayName("Suppress Null Values") + .description("Specifies how the writer should handle a null field") + .allowableValues(NEVER_SUPPRESS, ALWAYS_SUPPRESS, SUPPRESS_MISSING) + .defaultValue(NEVER_SUPPRESS.getValue()) + .required(true) + .build(); + private static final Set<Relationship> relationships; private static final List<PropertyDescriptor> propertyDescriptors; @@ -160,6 +179,8 @@ public class PutElasticsearchHttpRecord extends AbstractElasticsearchHttpProcess private final JsonFactory factory = new JsonFactory(); + private volatile String nullSuppression; + static { final Set<Relationship> _rels = new HashSet<>(); _rels.add(REL_SUCCESS); @@ -179,6 +200,7 @@ public class PutElasticsearchHttpRecord extends AbstractElasticsearchHttpProcess descriptors.add(INDEX); descriptors.add(TYPE); descriptors.add(INDEX_OP); + descriptors.add(SUPPRESS_NULLS); propertyDescriptors = Collections.unmodifiableList(descriptors); } @@ -284,6 +306,8 @@ public class PutElasticsearchHttpRecord extends AbstractElasticsearchHttpProcess return; } + 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 StringBuilder sb = new StringBuilder(); @@ -450,7 +474,10 @@ public class PutElasticsearchHttpRecord extends AbstractElasticsearchHttpProcess final String fieldName = field.getFieldName(); final Object value = record.getValue(field); if (value == null) { - generator.writeNullField(fieldName); + if (nullSuppression.equals(NEVER_SUPPRESS.getValue()) || (nullSuppression.equals(SUPPRESS_MISSING.getValue())) && record.getRawFieldNames().contains(fieldName)) { + generator.writeNullField(fieldName); + } + continue; } @@ -465,7 +492,10 @@ public class PutElasticsearchHttpRecord extends AbstractElasticsearchHttpProcess @SuppressWarnings("unchecked") private void writeValue(final JsonGenerator generator, final Object value, final String fieldName, final DataType dataType) throws IOException { if (value == null) { - generator.writeNull(); + if (nullSuppression.equals(NEVER_SUPPRESS.getValue()) || ((nullSuppression.equals(SUPPRESS_MISSING.getValue())) && fieldName != null && !fieldName.equals(""))) { + generator.writeNullField(fieldName); + } + return; } http://git-wip-us.apache.org/repos/asf/nifi/blob/6b34d3be/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecordIT.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecordIT.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecordIT.java new file mode 100644 index 0000000..0ff0f8e --- /dev/null +++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecordIT.java @@ -0,0 +1,211 @@ +/* + * 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 com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.nifi.serialization.SimpleRecordSchema; +import org.apache.nifi.serialization.record.MapRecord; +import org.apache.nifi.serialization.record.MockRecordParser; +import org.apache.nifi.serialization.record.RecordField; +import org.apache.nifi.serialization.record.RecordFieldType; +import org.apache.nifi.serialization.record.RecordSchema; +import org.apache.nifi.util.MockFlowFile; +import org.apache.nifi.util.TestRunner; +import org.apache.nifi.util.TestRunners; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public class PutElasticsearchHttpRecordIT { + protected TestRunner runner; + private MockRecordParser recordReader; + static RecordSchema personSchema; + static TestRunner FETCH_RUNNER; + ObjectMapper mapper = new ObjectMapper(); + + @BeforeClass + public static void setupTests() throws Exception { + final List<RecordField> personFields = new ArrayList<>(); + final RecordField nameField = new RecordField("name", RecordFieldType.STRING.getDataType()); + final RecordField ageField = new RecordField("age", RecordFieldType.INT.getDataType()); + final RecordField sportField = new RecordField("sport", RecordFieldType.STRING.getDataType()); + personFields.add(nameField); + personFields.add(ageField); + personFields.add(sportField); + personSchema = new SimpleRecordSchema(personFields); + + FETCH_RUNNER = TestRunners.newTestRunner(FetchElasticsearchHttp.class); + FETCH_RUNNER.setProperty(FetchElasticsearchHttp.ES_URL, "http://localhost:9200";); + FETCH_RUNNER.setProperty(FetchElasticsearchHttp.INDEX, "people_test"); + FETCH_RUNNER.setProperty(FetchElasticsearchHttp.TYPE, "person"); + FETCH_RUNNER.setProperty(FetchElasticsearchHttp.DOC_ID, "${doc_id}"); + FETCH_RUNNER.assertValid(); + } + + @Before + public void setup() throws Exception { + recordReader = new MockRecordParser(); + recordReader.addSchemaField("id", RecordFieldType.INT); + + recordReader.addSchemaField("person", RecordFieldType.RECORD); + + runner = TestRunners.newTestRunner(PutElasticsearchHttpRecord.class); + runner.addControllerService("reader", recordReader); + runner.enableControllerService(recordReader); + runner.setProperty(PutElasticsearchHttpRecord.RECORD_READER, "reader"); + runner.setProperty(PutElasticsearchHttpRecord.ES_URL, "http://localhost:9200";); + runner.setProperty(PutElasticsearchHttpRecord.INDEX, "people_test"); + runner.setProperty(PutElasticsearchHttpRecord.TYPE, "person"); + runner.setProperty(PutElasticsearchHttpRecord.ID_RECORD_PATH, "/id"); + runner.assertValid(); + } + + @After + public void tearDown() { + FETCH_RUNNER.clearTransferState(); + } + + private void setupPut() { + runner.enqueue(""); + runner.setValidateExpressionUsage(false); + runner.run(1, true, true); + runner.assertTransferCount(PutElasticsearchHttpRecord.REL_FAILURE, 0); + runner.assertTransferCount(PutElasticsearchHttpRecord.REL_RETRY, 0); + runner.assertTransferCount(PutElasticsearchHttpRecord.REL_SUCCESS, 1); + } + + private void testFetch(List<Map<String, String>> attrs) { + for (Map<String, String> attr : attrs) { + FETCH_RUNNER.enqueue("", attr); + } + + FETCH_RUNNER.run(attrs.size(), true, true); + FETCH_RUNNER.assertTransferCount(FetchElasticsearchHttp.REL_FAILURE, 0); + FETCH_RUNNER.assertTransferCount(FetchElasticsearchHttp.REL_RETRY, 0); + FETCH_RUNNER.assertTransferCount(FetchElasticsearchHttp.REL_NOT_FOUND, 0); + FETCH_RUNNER.assertTransferCount(FetchElasticsearchHttp.REL_SUCCESS, attrs.size()); + } + + @Test + public void testNoNullSuppresion() throws Exception { + recordReader.addRecord(1, new MapRecord(personSchema, new HashMap<String,Object>() {{ + put("name", "John Doe"); + put("age", 48); + put("sport", null); + }})); + + List<Map<String, String>> attrs = new ArrayList<>(); + Map<String, String> attr = new HashMap<>(); + attr.put("doc_id", "1"); + attrs.add(attr); + + setupPut(); + testFetch(attrs); + + byte[] raw = FETCH_RUNNER.getContentAsByteArray(FETCH_RUNNER.getFlowFilesForRelationship(FetchElasticsearchHttp.REL_SUCCESS).get(0)); + String val = new String(raw); + + Map<String, Object> parsed = mapper.readValue(val, Map.class); + Assert.assertNotNull(parsed); + Map<String, Object> person = (Map)parsed.get("person"); + Assert.assertNotNull(person); + Assert.assertTrue(person.containsKey("sport")); + Assert.assertNull(person.get("sport")); + } + + private void sharedSuppressTest(SharedPostTest spt) throws Exception { + List<Map<String, String>> attrs = new ArrayList<>(); + Map<String, String> attr = new HashMap<>(); + attr.put("doc_id", "1"); + attrs.add(attr); + attr = new HashMap<>(); + attr.put("doc_id", "2"); + attrs.add(attr); + + setupPut(); + testFetch(attrs); + + List<MockFlowFile> flowFiles = FETCH_RUNNER.getFlowFilesForRelationship(FetchElasticsearchHttp.REL_SUCCESS); + String ff1 = new String(FETCH_RUNNER.getContentAsByteArray(flowFiles.get(0))); + String ff2 = new String(FETCH_RUNNER.getContentAsByteArray(flowFiles.get(1))); + Map<String, Object> ff1Map = mapper.readValue(ff1, Map.class); + Map<String, Object> ff2Map = mapper.readValue(ff2, Map.class); + Assert.assertNotNull(ff1Map); + Assert.assertNotNull(ff2Map); + Map<String, Object> p1 = (Map)ff1Map.get("person"); + Map<String, Object> p2 = (Map)ff2Map.get("person"); + Assert.assertNotNull(p1); + Assert.assertNotNull(p2); + + spt.run(p1, p2); + } + + @Test + public void testMissingRecord() throws Exception { + recordReader.addRecord(1, new MapRecord(personSchema, new HashMap<String,Object>() {{ + put("name", "John Doe"); + put("age", 48); + }})); + + recordReader.addRecord(2, new MapRecord(personSchema, new HashMap<String,Object>() {{ + put("name", "John Doe"); + put("age", 48); + put("sport", null); + }})); + + runner.setProperty(PutElasticsearchHttpRecord.SUPPRESS_NULLS, PutElasticsearchHttpRecord.SUPPRESS_MISSING); + + sharedSuppressTest((p1, p2) -> { + Assert.assertFalse(p1.containsKey("sport")); + Assert.assertTrue(p2.containsKey("sport")); + Assert.assertNull(p2.get("sport")); + }); + } + + @Test + public void testAlwaysSuppress() throws Exception { + recordReader.addRecord(1, new MapRecord(personSchema, new HashMap<String,Object>() {{ + put("name", "John Doe"); + put("age", 48); + }})); + + recordReader.addRecord(2, new MapRecord(personSchema, new HashMap<String,Object>() {{ + put("name", "John Doe"); + put("age", 48); + put("sport", null); + }})); + + runner.setProperty(PutElasticsearchHttpRecord.SUPPRESS_NULLS, PutElasticsearchHttpRecord.ALWAYS_SUPPRESS); + + sharedSuppressTest((p1, p2) -> { + Assert.assertFalse(p1.containsKey("sport")); + Assert.assertFalse(p2.containsKey("sport")); + }); + } + + private interface SharedPostTest { + void run(Map<String, Object> p1, Map<String, Object> p2); + } +}
