Repository: nifi Updated Branches: refs/heads/master 748cf7456 -> b1478cdb1
NIFI-5051 Created ElasticSearch lookup service. NIFI-5051 Fixed checkstyle issue. NIFI-5051 Converted ES lookup service to use a SchemaRegistry. NIFI-5051 Cleaned up POM and added a simple unit test that uses a mock client service. NIFI-5051 Added change; waiting for feedback. NIFI-5051 Changed query setup based on code review. Changed tests to Groovy to make them easier to read with all of the inline JSON. NIFI-5051 fixed a checkstyle issue. NIFI-5051 Rebased to cleanup merge issues NIFI-5051 Added changes from a code review. NIFI-5051 Fixed a checkstyle issue. NIFI-5051 Added coverage generator for tests. Rebased. NIFI-5051 Updated service and switched it over to JsonInferenceSchemaRegistryService. NIFI-5051 Removed dead code. NIFI-5051 Fixed checkstyle errors. NIFI-5051 Refactored query builder. NIFI-5051 Added placeholder gitignore to force test compile. NIFI-5051 Added note explaining why the .gitignore file was needed. NIFI-5051 Made constructor public. NIFI-5051 Fixed path issue in client service integration tests. NIFI-5051 Added additional mapping capabilities to let users massage the result set into the fields they want. Signed-off-by: Matthew Burgess <[email protected]> This closes #2615 Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/b1478cdb Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/b1478cdb Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/b1478cdb Branch: refs/heads/master Commit: b1478cdb195722d7d285cad53d1f89c325868f55 Parents: 748cf74 Author: Mike Thomsen <[email protected]> Authored: Fri Apr 6 21:38:07 2018 -0400 Committer: Matthew Burgess <[email protected]> Committed: Mon Oct 1 10:48:45 2018 -0400 ---------------------------------------------------------------------- .../nifi-elasticsearch-client-service/pom.xml | 35 +++ .../ElasticSearchLookupService.java | 315 +++++++++++++++++++ ...org.apache.nifi.controller.ControllerService | 1 + .../additionalDetails.html | 53 ++++ .../ElasticSearch5ClientService_IT.groovy | 148 +++++++++ .../ElasticSearchLookupServiceTest.groovy | 69 ++++ .../ElasticSearchLookupService_IT.groovy | 211 +++++++++++++ .../TestControllerServiceProcessor.groovy | 51 +++ .../TestElasticSearchClientService.groovy | 75 +++++ .../integration/TestSchemaRegistry.groovy | 41 +++ .../src/test/java/.gitignore | 1 + .../ElasticSearchClientService_IT.java | 165 ---------- .../TestControllerServiceProcessor.java | 49 --- .../src/test/resources/setup.script | 11 +- 14 files changed, 1010 insertions(+), 215 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/b1478cdb/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/pom.xml ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/pom.xml b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/pom.xml index 8eb976d..b0c7842 100644 --- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/pom.xml +++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/pom.xml @@ -58,6 +58,12 @@ </dependency> <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-lookup-service-api</artifactId> + <scope>provided</scope> + </dependency> + + <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>2.9.5</version> @@ -123,6 +129,33 @@ </exclusion> </exclusions> </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-avro-record-utils</artifactId> + <version>1.8.0-SNAPSHOT</version> + <scope>compile</scope> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-schema-registry-service-api</artifactId> + <scope>compile</scope> + </dependency> + <dependency> + <groupId>com.jayway.jsonpath</groupId> + <artifactId>json-path</artifactId> + <version>2.4.0</version> + </dependency> + <dependency> + <groupId>org.mockito</groupId> + <artifactId>mockito-all</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-record-path</artifactId> + <version>1.8.0-SNAPSHOT</version> + <scope>compile</scope> + </dependency> </dependencies> <profiles> @@ -140,6 +173,8 @@ <httpPort>9400</httpPort> <version>5.6.2</version> <timeout>90</timeout> + <logLevel>ERROR</logLevel> + <pathInitScript>${project.basedir}/src/test/resources/setup.script</pathInitScript> </configuration> <executions> <execution> http://git-wip-us.apache.org/repos/asf/nifi/blob/b1478cdb/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchLookupService.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchLookupService.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchLookupService.java new file mode 100644 index 0000000..c864771 --- /dev/null +++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchLookupService.java @@ -0,0 +1,315 @@ +/* + * 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.elasticsearch; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.jayway.jsonpath.JsonPath; +import org.apache.nifi.annotation.lifecycle.OnEnabled; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.ValidationResult; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.lookup.LookupFailureException; +import org.apache.nifi.lookup.LookupService; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.record.path.FieldValue; +import org.apache.nifi.record.path.RecordPath; +import org.apache.nifi.schema.access.SchemaNotFoundException; +import org.apache.nifi.serialization.JsonInferenceSchemaRegistryService; +import org.apache.nifi.serialization.record.MapRecord; +import org.apache.nifi.serialization.record.Record; +import org.apache.nifi.serialization.record.RecordSchema; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; + +public class ElasticSearchLookupService extends JsonInferenceSchemaRegistryService implements LookupService<Record> { + public static final PropertyDescriptor CLIENT_SERVICE = new PropertyDescriptor.Builder() + .name("el-rest-client-service") + .displayName("Client Service") + .description("An ElasticSearch client service to use for running queries.") + .identifiesControllerService(ElasticSearchClientService.class) + .required(true) + .build(); + public static final PropertyDescriptor INDEX = new PropertyDescriptor.Builder() + .name("el-lookup-index") + .displayName("Index") + .description("The name of the index to read from") + .required(true) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor TYPE = new PropertyDescriptor.Builder() + .name("el-lookup-type") + .displayName("Type") + .description("The type of this document (used by Elasticsearch for indexing and searching)") + .required(false) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + private ElasticSearchClientService clientService; + + private String index; + private String type; + private ObjectMapper mapper; + + private final List<PropertyDescriptor> DESCRIPTORS; + + public ElasticSearchLookupService() { + List<PropertyDescriptor> _desc = new ArrayList<>(); + _desc.addAll(super.getSupportedPropertyDescriptors()); + _desc.add(CLIENT_SERVICE); + _desc.add(INDEX); + _desc.add(TYPE); + DESCRIPTORS = Collections.unmodifiableList(_desc); + } + + private volatile ConcurrentHashMap<String, RecordPath> mappings; + + @Override + @OnEnabled + public void onEnabled(final ConfigurationContext context) { + clientService = context.getProperty(CLIENT_SERVICE).asControllerService(ElasticSearchClientService.class); + index = context.getProperty(INDEX).evaluateAttributeExpressions().getValue(); + type = context.getProperty(TYPE).evaluateAttributeExpressions().getValue(); + mapper = new ObjectMapper(); + + List<PropertyDescriptor> dynamic = context.getProperties().entrySet().stream() + .filter( e -> e.getKey().isDynamic()) + .map(e -> e.getKey()) + .collect(Collectors.toList()); + + Map<String, RecordPath> _temp = new HashMap<>(); + for (PropertyDescriptor desc : dynamic) { + String value = context.getProperty(desc).getValue(); + String name = desc.getName(); + _temp.put(name, RecordPath.compile(value)); + } + + mappings = new ConcurrentHashMap<>(_temp); + + super.onEnabled(context); + } + + @Override + protected List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return DESCRIPTORS; + } + + @Override + public PropertyDescriptor getSupportedDynamicPropertyDescriptor(String name) { + return new PropertyDescriptor.Builder() + .name(name) + .addValidator((subject, input, context) -> { + ValidationResult.Builder builder = new ValidationResult.Builder(); + try { + JsonPath.parse(input); + builder.valid(true); + } catch (Exception ex) { + builder.explanation(ex.getMessage()) + .valid(false) + .subject(subject); + } + + return builder.build(); + }) + .dynamic(true) + .build(); + } + + @Override + public Optional<Record> lookup(Map<String, Object> coordinates) throws LookupFailureException { + Map<String, String> context = coordinates.entrySet().stream() + .collect(Collectors.toMap( + e -> e.getKey(), + e -> e.getValue().toString() + )); + return lookup(coordinates, context); + } + + @Override + public Optional<Record> lookup(Map<String, Object> coordinates, Map<String, String> context) throws LookupFailureException { + validateCoordinates(coordinates); + + try { + Record record; + if (coordinates.containsKey("_id")) { + record = getById((String)coordinates.get("_id"), context); + } else { + record = getByQuery(coordinates, context); + } + + return record == null ? Optional.empty() : Optional.of(record); + } catch (Exception ex) { + getLogger().error("Error during lookup.", ex); + throw new LookupFailureException(ex); + } + } + + private void validateCoordinates(Map coordinates) throws LookupFailureException { + List<String> reasons = new ArrayList<>(); + + if (coordinates.containsKey("_id") && !(coordinates.get("_id") instanceof String)) { + reasons.add("_id was supplied, but it was not a String."); + } + + if (coordinates.containsKey("_id") && coordinates.size() > 1) { + reasons.add("When _id is used, it can be the only key used in the lookup."); + } + + if (reasons.size() > 0) { + String error = String.join("\n", reasons); + throw new LookupFailureException(error); + } + } + + private Record getById(final String _id, Map<String, String> context) throws IOException, LookupFailureException, SchemaNotFoundException { + Map<String, Object> query = new HashMap<String, Object>(){{ + put("query", new HashMap<String, Object>() {{ + put("match", new HashMap<String, String>(){{ + put("_id", _id); + }}); + }}); + }}; + + String json = mapper.writeValueAsString(query); + + SearchResponse response = clientService.search(json, index, type); + + if (response.getNumberOfHits() > 1) { + throw new LookupFailureException(String.format("Expected 1 response, got %d for query %s", + response.getNumberOfHits(), json)); + } else if (response.getNumberOfHits() == 0) { + return null; + } + + final Map<String, Object> source = (Map)response.getHits().get(0).get("_source"); + + RecordSchema toUse = getSchema(context, source, null); + + Record record = new MapRecord(toUse, source); + + if (mappings.size() > 0) { + record = applyMappings(record, source); + } + + return record; + } + + Map<String, Object> getNested(String key, Object value) { + String path = key.substring(0, key.lastIndexOf(".")); + + return new HashMap<String, Object>(){{ + put("path", path); + put("query", new HashMap<String, Object>(){{ + put("match", new HashMap<String, Object>(){{ + put(key, value); + }}); + }}); + }}; + } + + private Map<String, Object> buildQuery(Map<String, Object> coordinates) { + Map<String, Object> query = new HashMap<String, Object>(){{ + put("bool", new HashMap<String, Object>(){{ + put("must", coordinates.entrySet().stream() + .map(e -> new HashMap<String, Object>(){{ + if (e.getKey().contains(".")) { + put("nested", getNested(e.getKey(), e.getValue())); + } else { + put("match", new HashMap<String, Object>() {{ + put(e.getKey(), e.getValue()); + }}); + } + }}).collect(Collectors.toList()) + ); + }}); + }}; + + Map<String, Object> outter = new HashMap<String, Object>(){{ + put("size", 1); + put("query", query); + }}; + + return outter; + } + + private Record getByQuery(final Map<String, Object> query, Map<String, String> context) throws LookupFailureException { + try { + final String json = mapper.writeValueAsString(buildQuery(query)); + + SearchResponse response = clientService.search(json, index, type); + + if (response.getNumberOfHits() == 0) { + return null; + } else { + final Map<String, Object> source = (Map)response.getHits().get(0).get("_source"); + RecordSchema toUse = getSchema(context, source, null); + Record record = new MapRecord(toUse, source); + + if (mappings.size() > 0) { + record = applyMappings(record, source); + } + + return record; + } + + } catch (Exception e) { + throw new LookupFailureException(e); + } + } + + private Record applyMappings(Record record, Map<String, Object> source) { + Record _rec = new MapRecord(record.getSchema(), new HashMap<>()); + + mappings.entrySet().forEach(entry -> { + try { + Object o = JsonPath.read(source, entry.getKey()); + RecordPath path = entry.getValue(); + Optional<FieldValue> first = path.evaluate(_rec).getSelectedFields().findFirst(); + if (first.isPresent()) { + first.get().updateValue(o); + } + } catch (Exception ex) { + throw new RuntimeException(ex); + } + }); + + return _rec; + } + + @Override + public Class<?> getValueType() { + return Record.class; + } + + @Override + public Set<String> getRequiredKeys() { + return Collections.emptySet(); + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/b1478cdb/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService index 161f652..65745fb 100644 --- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService +++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService @@ -12,4 +12,5 @@ # 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. +org.apache.nifi.elasticsearch.ElasticSearchLookupService org.apache.nifi.elasticsearch.ElasticSearchClientServiceImpl \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi/blob/b1478cdb/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/resources/docs/org.apache.nifi.elasticsearch.ElasticSearchLookupService/additionalDetails.html ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/resources/docs/org.apache.nifi.elasticsearch.ElasticSearchLookupService/additionalDetails.html b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/resources/docs/org.apache.nifi.elasticsearch.ElasticSearchLookupService/additionalDetails.html new file mode 100644 index 0000000..3b95430 --- /dev/null +++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/resources/docs/org.apache.nifi.elasticsearch.ElasticSearchLookupService/additionalDetails.html @@ -0,0 +1,53 @@ +<!DOCTYPE html> +<html lang="en"> +<!-- + 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. +--> +<head> + <meta charset="utf-8" /> + <title>ElasticSearchLookupService</title> + <link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css" /> +</head> + +<body> +<!-- Processor Documentation ================================================== --> +<h2>Description:</h2> +<p> + This lookup service uses ElasticSearch as its data source. Mappings in LookupRecord map record paths to paths within + an ElasticSearch document. Example: +</p> +<p> + <em>/user/name</em> => <em>user.contact.name</em> +</p> +<p> + That would map the record path <em>/user/name</em> to an embedded document named <em>contact</em> with a field named + <em>name</em>. +</p> +<p> + The query that is assembled from these is a boolean query where all of the criteria are under the <em>must</em> list. + In addition, wildcards are not supported right now and all criteria are translated into literal <em>match</em> queries. +</p> +<h2>Post-Processing</h2> +<p> + Because an ElasticSearch result might be structured differently than the record which will be enriched by this service, + users can specify an additional set of mappings on this lookup service that map JsonPath operations to record paths. Example: +</p> +<p> + <em>$.user.contact.email</em> => <em>/user/email_address</em> +</p> +<p> + Would copy the field <em>email</em> from the embedded document <em>contact</em> into the record at that path. +</p> +</body> +</html> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi/blob/b1478cdb/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/ElasticSearch5ClientService_IT.groovy ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/ElasticSearch5ClientService_IT.groovy b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/ElasticSearch5ClientService_IT.groovy new file mode 100644 index 0000000..b5c4468 --- /dev/null +++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/ElasticSearch5ClientService_IT.groovy @@ -0,0 +1,148 @@ +/* + * 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.elasticsearch.integration + +import org.apache.nifi.elasticsearch.DeleteOperationResponse +import org.apache.nifi.elasticsearch.ElasticSearchClientService +import org.apache.nifi.elasticsearch.ElasticSearchClientServiceImpl +import org.apache.nifi.elasticsearch.SearchResponse +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.Test + +import static groovy.json.JsonOutput.prettyPrint +import static groovy.json.JsonOutput.toJson + +class ElasticSearch5ClientService_IT { + + private TestRunner runner + private ElasticSearchClientServiceImpl service + + static final String INDEX = "messages" + static final String TYPE = "message" + + @Before + void before() throws Exception { + runner = TestRunners.newTestRunner(TestControllerServiceProcessor.class) + service = new ElasticSearchClientServiceImpl() + runner.addControllerService("Client Service", service) + runner.setProperty(service, ElasticSearchClientService.HTTP_HOSTS, "http://localhost:9400";) + runner.setProperty(service, ElasticSearchClientService.CONNECT_TIMEOUT, "10000") + runner.setProperty(service, ElasticSearchClientService.SOCKET_TIMEOUT, "60000") + runner.setProperty(service, ElasticSearchClientService.RETRY_TIMEOUT, "60000") + try { + runner.enableControllerService(service) + } catch (Exception ex) { + ex.printStackTrace() + throw ex + } + } + + @After + void after() throws Exception { + service.onDisabled() + } + + @Test + void testBasicSearch() throws Exception { + String query = prettyPrint(toJson([ + size: 10, + query: [ + match_all: [:] + ], + aggs: [ + term_counts: [ + terms: [ + field: "msg", + size: 5 + ] + ] + ] + ])) + + + SearchResponse response = service.search(query, "messages", "message") + Assert.assertNotNull("Response was null", response) + + Assert.assertEquals("Wrong count", 15, response.numberOfHits) + Assert.assertFalse("Timed out", response.isTimedOut()) + Assert.assertNotNull("Hits was null", response.getHits()) + Assert.assertEquals("Wrong number of hits", 10, response.hits.size()) + Assert.assertNotNull("Aggregations are missing", response.aggregations) + Assert.assertEquals("Aggregation count is wrong", 1, response.aggregations.size()) + + Map termCounts = response.aggregations.get("term_counts") + Assert.assertNotNull("Term counts was missing", termCounts) + def buckets = termCounts.get("buckets") + Assert.assertNotNull("Buckets branch was empty", buckets) + def expected = [ + "one": 1, + "two": 2, + "three": 3, + "four": 4, + "five": 5 + ] + + buckets.each { aggRes -> + def key = aggRes["key"] + def docCount = aggRes["doc_count"] + Assert.assertEquals("${key} did not match.", expected[key], docCount) + } + } + + @Test + void testDeleteByQuery() throws Exception { + String query = prettyPrint(toJson([ + query: [ + match: [ + msg: "five" + ] + ] + ])) + DeleteOperationResponse response = service.deleteByQuery(query, INDEX, TYPE) + Assert.assertNotNull(response) + Assert.assertTrue(response.getTook() > 0) + } + + @Test + void testDeleteById() throws Exception { + final String ID = "1" + DeleteOperationResponse response = service.deleteById(INDEX, TYPE, ID) + Assert.assertNotNull(response) + Assert.assertTrue(response.getTook() > 0) + def doc = service.get(INDEX, TYPE, ID) + Assert.assertNull(doc) + doc = service.get(INDEX, TYPE, "2") + Assert.assertNotNull(doc) + } + + @Test + void testGet() throws IOException { + Map old + 1.upto(15) { index -> + String id = String.valueOf(index) + def doc = service.get(INDEX, TYPE, id) + Assert.assertNotNull("Doc was null", doc) + Assert.assertNotNull("${doc.toString()}\t${doc.keySet().toString()}", doc.get("msg")) + old = doc + } + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/b1478cdb/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/ElasticSearchLookupServiceTest.groovy ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/ElasticSearchLookupServiceTest.groovy b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/ElasticSearchLookupServiceTest.groovy new file mode 100644 index 0000000..27452a3 --- /dev/null +++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/ElasticSearchLookupServiceTest.groovy @@ -0,0 +1,69 @@ +/* + * 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.elasticsearch.integration + +import org.apache.nifi.elasticsearch.ElasticSearchClientService +import org.apache.nifi.elasticsearch.ElasticSearchLookupService +import org.apache.nifi.schema.access.SchemaAccessUtils +import org.apache.nifi.serialization.record.MapRecord +import org.apache.nifi.util.TestRunner +import org.apache.nifi.util.TestRunners +import org.junit.Assert +import org.junit.Before +import org.junit.Test + +class ElasticSearchLookupServiceTest { + ElasticSearchClientService mockClientService + ElasticSearchLookupService lookupService + TestRunner runner + + @Before + void setup() throws Exception { + mockClientService = new TestElasticSearchClientService() + lookupService = new ElasticSearchLookupService() + def registry = new TestSchemaRegistry() + runner = TestRunners.newTestRunner(TestControllerServiceProcessor.class) + runner.addControllerService("clientService", mockClientService) + runner.addControllerService("lookupService", lookupService) + runner.addControllerService("registry", registry) + runner.enableControllerService(mockClientService) + runner.enableControllerService(registry) + runner.setProperty(lookupService, ElasticSearchLookupService.CLIENT_SERVICE, "clientService") + runner.setProperty(lookupService, ElasticSearchLookupService.INDEX, "users") + runner.setProperty(TestControllerServiceProcessor.CLIENT_SERVICE, "clientService") + runner.setProperty(TestControllerServiceProcessor.LOOKUP_SERVICE, "lookupService") + runner.setProperty(lookupService, SchemaAccessUtils.SCHEMA_REGISTRY, "registry") + runner.setProperty(lookupService, SchemaAccessUtils.SCHEMA_ACCESS_STRATEGY, SchemaAccessUtils.INFER_SCHEMA) + runner.enableControllerService(lookupService) + } + + @Test + void simpleLookupTest() throws Exception { + def coordinates = ["_id": "12345" ] + + Optional<MapRecord> result = lookupService.lookup(coordinates) + + Assert.assertNotNull(result) + Assert.assertTrue(result.isPresent()) + MapRecord record = result.get() + Assert.assertEquals("john.smith", record.getAsString("username")) + Assert.assertEquals("testing1234", record.getAsString("password")) + Assert.assertEquals("[email protected]", record.getAsString("email")) + Assert.assertEquals("Software Engineer", record.getAsString("position")) + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/b1478cdb/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/ElasticSearchLookupService_IT.groovy ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/ElasticSearchLookupService_IT.groovy b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/ElasticSearchLookupService_IT.groovy new file mode 100644 index 0000000..b255836 --- /dev/null +++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/ElasticSearchLookupService_IT.groovy @@ -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.elasticsearch.integration + +import org.apache.nifi.elasticsearch.ElasticSearchClientService +import org.apache.nifi.elasticsearch.ElasticSearchClientServiceImpl +import org.apache.nifi.elasticsearch.ElasticSearchLookupService +import org.apache.nifi.lookup.LookupFailureException +import org.apache.nifi.record.path.RecordPath +import org.apache.nifi.schema.access.SchemaAccessUtils +import org.apache.nifi.schemaregistry.services.SchemaRegistry +import org.apache.nifi.serialization.record.MapRecord +import org.apache.nifi.serialization.record.Record +import org.apache.nifi.serialization.record.RecordSchema +import org.apache.nifi.serialization.record.type.RecordDataType +import org.apache.nifi.util.TestRunner +import org.apache.nifi.util.TestRunners +import org.junit.Assert +import org.junit.Before +import org.junit.Test + +class ElasticSearchLookupService_IT { + private TestRunner runner + private ElasticSearchClientService service + private ElasticSearchLookupService lookupService + + @Before + void before() throws Exception { + runner = TestRunners.newTestRunner(TestControllerServiceProcessor.class) + service = new ElasticSearchClientServiceImpl() + lookupService = new ElasticSearchLookupService() + runner.addControllerService("Client Service", service) + runner.addControllerService("Lookup Service", lookupService) + runner.setProperty(service, ElasticSearchClientService.HTTP_HOSTS, "http://localhost:9400";) + runner.setProperty(service, ElasticSearchClientService.CONNECT_TIMEOUT, "10000") + runner.setProperty(service, ElasticSearchClientService.SOCKET_TIMEOUT, "60000") + runner.setProperty(service, ElasticSearchClientService.RETRY_TIMEOUT, "60000") + runner.setProperty(TestControllerServiceProcessor.CLIENT_SERVICE, "Client Service") + runner.setProperty(TestControllerServiceProcessor.LOOKUP_SERVICE, "Lookup Service") + runner.setProperty(lookupService, ElasticSearchLookupService.CLIENT_SERVICE, "Client Service") + runner.setProperty(lookupService, ElasticSearchLookupService.INDEX, "user_details") + runner.setProperty(lookupService, ElasticSearchLookupService.TYPE, "details") + + try { + runner.enableControllerService(service) + runner.enableControllerService(lookupService) + } catch (Exception ex) { + ex.printStackTrace() + throw ex + } + } + + @Test + void testValidity() throws Exception { + setDefaultSchema() + runner.assertValid() + } + + private void setDefaultSchema() throws Exception { + runner.disableControllerService(lookupService) + SchemaRegistry registry = new TestSchemaRegistry() + runner.addControllerService("registry", registry) + runner.setProperty(lookupService, SchemaAccessUtils.SCHEMA_REGISTRY, "registry") + runner.enableControllerService(registry) + runner.enableControllerService(lookupService) + } + + @Test + void lookupById() { + def coordinates = [ _id: "2" ] + Optional<Record> result = lookupService.lookup(coordinates) + + Assert.assertNotNull(result) + Assert.assertTrue(result.isPresent()) + def record = result.get() + Assert.assertEquals("[email protected]", record.getAsString("email")) + Assert.assertEquals("098-765-4321", record.getAsString("phone")) + Assert.assertEquals("GHIJK", record.getAsString("accessKey")) + } + + @Test + void testInvalidIdScenarios() { + def coordinates = [ + [ + _id: 1 + ], + [ + _id: "1", "email": "[email protected]" + ] + ] + + coordinates.each { coordinate -> + def exception + + try { + lookupService.lookup(coordinate) + } catch (Exception ex) { + exception = ex + } + + Assert.assertNotNull(exception) + Assert.assertTrue(exception instanceof LookupFailureException) + } + } + + @Test + void lookupByQuery() { + def coordinates = [ "phone": "098-765-4321", "email": "[email protected]" ] + Optional<Record> result = lookupService.lookup(coordinates) + + Assert.assertNotNull(result) + Assert.assertTrue(result.isPresent()) + def record = result.get() + Assert.assertEquals("[email protected]", record.getAsString("email")) + Assert.assertEquals("098-765-4321", record.getAsString("phone")) + Assert.assertEquals("GHIJK", record.getAsString("accessKey")) + } + + @Test + void testNestedSchema() { + def coordinates = [ + "subField.deeper.deepest.super_secret": "The sky is blue", + "subField.deeper.secretz": "Buongiorno, mondo!!", + "msg": "Hello, world" + ] + + runner.disableControllerService(lookupService) + runner.setProperty(lookupService, ElasticSearchLookupService.INDEX, "nested") + runner.setProperty(lookupService, ElasticSearchLookupService.TYPE, "nested_complex") + runner.enableControllerService(lookupService) + + Optional<Record> response = lookupService.lookup(coordinates) + Assert.assertNotNull(response) + Assert.assertTrue(response.isPresent()) + def rec = response.get() + Assert.assertEquals("Hello, world", rec.getValue("msg")) + def subRec = getSubRecord(rec, "subField") + Assert.assertNotNull(subRec) + def deeper = getSubRecord(subRec, "deeper") + Assert.assertNotNull(deeper) + def deepest = getSubRecord(deeper, "deepest") + Assert.assertNotNull(deepest) + Assert.assertEquals("The sky is blue", deepest.getAsString("super_secret")) + } + + @Test + void testDetectedSchema() throws LookupFailureException { + runner.disableControllerService(lookupService) + runner.setProperty(lookupService, ElasticSearchLookupService.INDEX, "complex") + runner.setProperty(lookupService, ElasticSearchLookupService.TYPE, "complex") + runner.enableControllerService(lookupService) + def coordinates = ["_id": "1" ] + + Optional<Record> response = lookupService.lookup(coordinates) + Assert.assertNotNull(response) + Record rec = response.get() + Record subRec = getSubRecord(rec, "subField") + + def r2 = new MapRecord(rec.schema, [:]) + def path = RecordPath.compile("/subField/longField") + def result = path.evaluate(r2) + result.selectedFields.findFirst().get().updateValue(1234567890L) + + Assert.assertNotNull(rec) + Assert.assertNotNull(subRec) + Assert.assertEquals("Hello, world", rec.getValue("msg")) + Assert.assertNotNull(rec.getValue("subField")) + Assert.assertEquals(new Long(100000), subRec.getValue("longField")) + Assert.assertEquals("2018-04-10T12:18:05Z", subRec.getValue("dateField")) + } + + Record getSubRecord(Record rec, String fieldName) { + RecordSchema schema = rec.schema + RecordSchema subSchema = ((RecordDataType)schema.getField(fieldName).get().dataType).childSchema + rec.getAsRecord(fieldName, subSchema) + } + + @Test + void testMappings() { + runner.disableControllerService(lookupService) + runner.setProperty(lookupService, "\$.subField.longField", "/longField2") + runner.setProperty(lookupService, '$.subField.dateField', '/dateField2') + runner.setProperty(lookupService, ElasticSearchLookupService.INDEX, "nested") + runner.setProperty(lookupService, ElasticSearchLookupService.TYPE, "nested_complex") + runner.enableControllerService(lookupService) + + def coordinates = ["msg": "Hello, world"] + def result = lookupService.lookup(coordinates) + Assert.assertTrue(result.isPresent()) + def rec = result.get() + ["dateField": "2018-08-14T10:08:00Z", "longField": 150000L].each { field -> + def value = rec.getValue(field.key) + Assert.assertEquals(field.value, value) + } + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/b1478cdb/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/TestControllerServiceProcessor.groovy ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/TestControllerServiceProcessor.groovy b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/TestControllerServiceProcessor.groovy new file mode 100644 index 0000000..75e9dbd --- /dev/null +++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/TestControllerServiceProcessor.groovy @@ -0,0 +1,51 @@ +/* + * 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.elasticsearch.integration + +import org.apache.nifi.components.PropertyDescriptor +import org.apache.nifi.elasticsearch.ElasticSearchClientServiceImpl +import org.apache.nifi.elasticsearch.ElasticSearchLookupService +import org.apache.nifi.processor.AbstractProcessor +import org.apache.nifi.processor.ProcessContext +import org.apache.nifi.processor.ProcessSession +import org.apache.nifi.processor.exception.ProcessException + +class TestControllerServiceProcessor extends AbstractProcessor { + + public static final PropertyDescriptor CLIENT_SERVICE = new PropertyDescriptor.Builder() + .name("Client Service") + .description("ElasticSearchClientServiceImpl") + .identifiesControllerService(ElasticSearchClientServiceImpl.class) + .required(true) + .build() + public static final PropertyDescriptor LOOKUP_SERVICE = new PropertyDescriptor.Builder() + .name("Lookup Service") + .description("ElasticSearchClientServiceImpl") + .identifiesControllerService(ElasticSearchLookupService.class) + .required(false) + .build() + + @Override + void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException { + } + + @Override + protected List<PropertyDescriptor> getSupportedPropertyDescriptors() { + [ CLIENT_SERVICE, LOOKUP_SERVICE ] + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/b1478cdb/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/TestElasticSearchClientService.groovy ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/TestElasticSearchClientService.groovy b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/TestElasticSearchClientService.groovy new file mode 100644 index 0000000..3b5fc0a --- /dev/null +++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/TestElasticSearchClientService.groovy @@ -0,0 +1,75 @@ +/* + * 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.elasticsearch.integration + +import org.apache.nifi.controller.AbstractControllerService +import org.apache.nifi.elasticsearch.DeleteOperationResponse +import org.apache.nifi.elasticsearch.ElasticSearchClientService +import org.apache.nifi.elasticsearch.IndexOperationRequest +import org.apache.nifi.elasticsearch.IndexOperationResponse +import org.apache.nifi.elasticsearch.SearchResponse + +class TestElasticSearchClientService extends AbstractControllerService implements ElasticSearchClientService { + @Override + IndexOperationResponse add(IndexOperationRequest operation) throws IOException { + return null + } + + @Override + IndexOperationResponse add(List<IndexOperationRequest> operations) throws IOException { + return null + } + + @Override + DeleteOperationResponse deleteById(String index, String type, String id) throws IOException { + return null + } + + @Override + DeleteOperationResponse deleteById(String index, String type, List<String> ids) throws IOException { + return null + } + + @Override + DeleteOperationResponse deleteByQuery(String query, String index, String type) throws IOException { + return null + } + + @Override + Map<String, Object> get(String index, String type, String id) throws IOException { + return null + } + + @Override + SearchResponse search(String query, String index, String type) throws IOException { + List hits = [[ + "_source": [ + "username": "john.smith", + "password": "testing1234", + "email": "[email protected]", + "position": "Software Engineer" + ] + ]] + return new SearchResponse(hits, null, 1, 100, false) + } + + @Override + String getTransitUrl(String index, String type) { + return "" + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/b1478cdb/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/TestSchemaRegistry.groovy ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/TestSchemaRegistry.groovy b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/TestSchemaRegistry.groovy new file mode 100644 index 0000000..2a0bd8e --- /dev/null +++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/groovy/org/apache/nifi/elasticsearch/integration/TestSchemaRegistry.groovy @@ -0,0 +1,41 @@ +/* + * 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.elasticsearch.integration + +import org.apache.nifi.controller.AbstractControllerService +import org.apache.nifi.schema.access.SchemaField +import org.apache.nifi.schemaregistry.services.SchemaRegistry +import org.apache.nifi.serialization.SimpleRecordSchema +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.serialization.record.SchemaIdentifier + +class TestSchemaRegistry extends AbstractControllerService implements SchemaRegistry { + @Override + RecordSchema retrieveSchema(SchemaIdentifier schemaIdentifier) { + new SimpleRecordSchema([ + new RecordField("msg", RecordFieldType.STRING.dataType) + ]) + } + + @Override + Set<SchemaField> getSuppliedSchemaFields() { + [ SchemaField.SCHEMA_NAME ] + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/b1478cdb/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/.gitignore ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/.gitignore b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/.gitignore new file mode 100644 index 0000000..00aca0c --- /dev/null +++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/.gitignore @@ -0,0 +1 @@ +# This is a placeholder to force Maven to compile the groovy code. \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi/blob/b1478cdb/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/ElasticSearchClientService_IT.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/ElasticSearchClientService_IT.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/ElasticSearchClientService_IT.java deleted file mode 100644 index 687faf0..0000000 --- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/ElasticSearchClientService_IT.java +++ /dev/null @@ -1,165 +0,0 @@ -/* - * 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.elasticsearch.integration; - -import org.apache.nifi.elasticsearch.DeleteOperationResponse; -import org.apache.nifi.elasticsearch.ElasticSearchClientService; -import org.apache.nifi.elasticsearch.ElasticSearchClientServiceImpl; -import org.apache.nifi.elasticsearch.IndexOperationRequest; -import org.apache.nifi.elasticsearch.SearchResponse; -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.Test; - -import java.io.IOException; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -public class ElasticSearchClientService_IT { - - private TestRunner runner; - private ElasticSearchClientServiceImpl service; - - static final String INDEX = "messages"; - static final String TYPE = "message"; - - @Before - public void before() throws Exception { - runner = TestRunners.newTestRunner(TestControllerServiceProcessor.class); - service = new ElasticSearchClientServiceImpl(); - runner.addControllerService("Client Service", service); - runner.setProperty(service, ElasticSearchClientService.HTTP_HOSTS, "http://localhost:9400";); - runner.setProperty(service, ElasticSearchClientService.CONNECT_TIMEOUT, "10000"); - runner.setProperty(service, ElasticSearchClientService.SOCKET_TIMEOUT, "60000"); - runner.setProperty(service, ElasticSearchClientService.RETRY_TIMEOUT, "60000"); - try { - runner.enableControllerService(service); - } catch (Exception ex) { - ex.printStackTrace(); - throw ex; - } - - Map<String, Integer> expected = new HashMap<>(); - expected.put("one", 1); - expected.put("two", 2); - expected.put("three", 3); - expected.put("four", 4); - expected.put("five", 5); - - - int index = 1; - List<IndexOperationRequest> docs = new ArrayList<>(); - for (Map.Entry<String, Integer> entry : expected.entrySet()) { - for (int idx = 0; idx < entry.getValue(); idx++) { - Map<String, Object> fields = new HashMap<>(); - fields.put("msg", entry.getKey()); - IndexOperationRequest ior = new IndexOperationRequest(INDEX, TYPE, String.valueOf(index++), fields); - docs.add(ior); - } - } - service.add(docs); - } - - @After - public void after() throws Exception { - service.onDisabled(); - } - - @Test - public void testBasicSearch() throws Exception { - String query = "{\n" + - "\t\"size\": 10,\n" + - "\t\"query\": {\n" + - "\t\t\"match_all\": {}\n" + - "\t},\n" + - "\t\"aggs\": {\n" + - "\t\t\"term_counts\": {\n" + - "\t\t\t\"terms\": {\n" + - "\t\t\t\t\"field\": \"msg.keyword\",\n" + - "\t\t\t\t\"size\": 5\n" + - "\t\t\t}\n" + - "\t\t}\n" + - "\t}\n" + - "}"; - SearchResponse response = service.search(query, INDEX, TYPE); - Assert.assertNotNull("Response was null", response); - - Assert.assertEquals("Wrong count", 15, response.getNumberOfHits()); - Assert.assertFalse("Timed out", response.isTimedOut()); - Assert.assertNotNull("Hits was null", response.getHits()); - Assert.assertEquals("Wrong number of hits", 10, response.getHits().size()); - Assert.assertNotNull("Aggregations are missing", response.getAggregations()); - Assert.assertEquals("Aggregation count is wrong", 1, response.getAggregations().size()); - - Map<String, Object> termCounts = (Map<String, Object>) response.getAggregations().get("term_counts"); - Assert.assertNotNull("Term counts was missing", termCounts); - List<Map<String, Object>> buckets = (List<Map<String, Object>>) termCounts.get("buckets"); - Assert.assertNotNull("Buckets branch was empty", buckets); - Map<String, Integer> expected = new HashMap<>(); - expected.put("one", 1); - expected.put("two", 2); - expected.put("three", 3); - expected.put("four", 4); - expected.put("five", 5); - - for (Map<String, Object> aggRes : buckets) { - String key = (String)aggRes.get("key"); - Integer docCount = (Integer)aggRes.get("doc_count"); - - Assert.assertEquals(String.format("%s did not match", key), expected.get(key), docCount); - } - } - - @Test - public void testDeleteByQuery() throws Exception { - String query = "{\"query\":{\"match\":{\"msg\":\"five\"}}}"; - DeleteOperationResponse response = service.deleteByQuery(query, INDEX, TYPE); - Assert.assertNotNull(response); - Assert.assertTrue(response.getTook() > 0); - } - - @Test - public void testDeleteById() throws Exception { - final String ID = "1"; - DeleteOperationResponse response = service.deleteById(INDEX, TYPE, ID); - Assert.assertNotNull(response); - Assert.assertTrue(response.getTook() > 0); - Map<String, Object> doc = service.get(INDEX, TYPE, ID); - Assert.assertNull(doc); - doc = service.get(INDEX, TYPE, "2"); - Assert.assertNotNull(doc); - } - - @Test - public void testGet() throws IOException { - Map<String, Object> old = null; - for (int index = 1; index <= 15; index++) { - String id = String.valueOf(index); - Map<String, Object> doc = service.get(INDEX, TYPE, id); - Assert.assertNotNull(doc); - Assert.assertNotNull(doc.toString() + "\t" + doc.keySet().toString(), doc.get("msg")); - Assert.assertFalse(doc == old); - old = doc; - } - } -} http://git-wip-us.apache.org/repos/asf/nifi/blob/b1478cdb/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/TestControllerServiceProcessor.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/TestControllerServiceProcessor.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/TestControllerServiceProcessor.java deleted file mode 100644 index 674cc14..0000000 --- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/TestControllerServiceProcessor.java +++ /dev/null @@ -1,49 +0,0 @@ -/* - * 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.elasticsearch.integration; - -import org.apache.nifi.components.PropertyDescriptor; -import org.apache.nifi.elasticsearch.ElasticSearchClientServiceImpl; -import org.apache.nifi.processor.AbstractProcessor; -import org.apache.nifi.processor.ProcessContext; -import org.apache.nifi.processor.ProcessSession; -import org.apache.nifi.processor.exception.ProcessException; - -import java.util.ArrayList; -import java.util.List; - -public class TestControllerServiceProcessor extends AbstractProcessor { - - static final PropertyDescriptor CLIENT_SERVICE = new PropertyDescriptor.Builder() - .name("Client Service") - .description("ElasticSearchClientServiceImpl") - .identifiesControllerService(ElasticSearchClientServiceImpl.class) - .required(true) - .build(); - - @Override - public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException { - } - - @Override - protected List<PropertyDescriptor> getSupportedPropertyDescriptors() { - List<PropertyDescriptor> propDescs = new ArrayList<>(); - propDescs.add(CLIENT_SERVICE); - return propDescs; - } -} http://git-wip-us.apache.org/repos/asf/nifi/blob/b1478cdb/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/resources/setup.script ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/resources/setup.script b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/resources/setup.script index 8cf4c97..a17a039 100644 --- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/resources/setup.script +++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/resources/setup.script @@ -13,7 +13,10 @@ # See the License for the specific language governing permissions and # limitations under the License. #create mapping +PUT:user_details/:{ "mappings":{"details":{ "properties":{ "email":{"type":"keyword"},"phone":{"type": "keyword"},"accessKey":{"type": "keyword"}}}}} PUT:messages/:{ "mappings":{"message":{ "properties":{ "msg":{"type":"keyword"}}}}} +PUT:complex/:{"mappings":{"complex":{"properties":{"msg":{"type":"keyword"},"subField":{"type":"nested","properties":{"longField":{"type":"long"},"dateField":{"type":"date"}}}}}}} +PUT:nested/:{"mappings":{"nested_complex":{"properties":{"msg":{"type":"keyword"},"subField":{"type":"nested","properties":{"longField":{"type":"long"},"dateField":{"type":"date"},"deeper":{"type":"nested","properties":{"secretz":{"type":"keyword"},"deepest":{"type":"nested","properties":{"super_secret":{"type":"keyword"}}}}}}}}}}} #add document PUT:messages/message/1:{ "msg":"one" } PUT:messages/message/2:{ "msg":"two" } @@ -29,4 +32,10 @@ PUT:messages/message/11:{ "msg":"five" } PUT:messages/message/12:{ "msg":"five" } PUT:messages/message/13:{ "msg":"five" } PUT:messages/message/14:{ "msg":"five" } -PUT:messages/message/15:{ "msg":"five" } \ No newline at end of file +PUT:messages/message/15:{ "msg":"five" } +PUT:complex/complex/1:{"msg":"Hello, world","subField":{"longField":100000,"dateField":"2018-04-10T12:18:05Z"}} +PUT:user_details/details/1:{ "email": "[email protected]", "phone": "123-456-7890", "accessKey": "ABCDE"} +PUT:user_details/details/2:{ "email": "[email protected]", "phone": "098-765-4321", "accessKey": "GHIJK"} +PUT:nested/nested_complex/1:{"msg":"Hello, world","subField":{"longField":150000,"dateField":"2018-08-14T10:08:00Z","deeper":{"secretz":"No one should see this!","deepest":{"super_secret":"Got nothin to hide"}}}} +PUT:nested/nested_complex/2:{"msg":"Hello, world","subField":{"longField":150000,"dateField":"2018-08-14T10:08:00Z","deeper":{"secretz":"Hello, world!","deepest":{"super_secret":"I could tell, but then I would have to kill you"}}}} +PUT:nested/nested_complex/3:{"msg":"Hello, world","subField":{"longField":150000,"dateField":"2018-08-14T10:08:00Z","deeper":{"secretz":"Buongiorno, mondo!!","deepest":{"super_secret":"The sky is blue"}}}} \ No newline at end of file
