You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2018/05/23 13:16:26 UTC
nifi git commit: NIFI-5052 Added DeleteByQuery ElasticSearch
processor.
Repository: nifi
Updated Branches:
refs/heads/master f7035f049 -> 3a29c1e4c
NIFI-5052 Added DeleteByQuery ElasticSearch processor.
NIFI-5052 fixed checkstyle issue.
NIFI-5052 Added changes requested in a code review.
NIFI-5052 Fixed a typo.
NIFI-5052 Added changes requested in a code review.
Signed-off-by: Matthew Burgess <ma...@apache.org>
This closes #2616
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/3a29c1e4
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/3a29c1e4
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/3a29c1e4
Branch: refs/heads/master
Commit: 3a29c1e4ca703a8d042438cd7b576485371dacbd
Parents: f7035f0
Author: Mike Thomsen <mi...@gmail.com>
Authored: Sat Apr 7 06:03:01 2018 -0400
Committer: Matthew Burgess <ma...@apache.org>
Committed: Wed May 23 09:14:04 2018 -0400
----------------------------------------------------------------------
.../elasticsearch/DeleteOperationResponse.java | 30 ++++
.../ElasticSearchClientService.java | 62 +++++++
.../elasticsearch/IndexOperationRequest.java | 50 ++++++
.../elasticsearch/IndexOperationResponse.java | 36 ++++
.../nifi-elasticsearch-client-service/pom.xml | 18 +-
.../ElasticSearchClientServiceImpl.java | 83 +++++++++-
.../ElasticSearch5ClientService_IT.java | 105 ------------
.../ElasticSearchClientService_IT.java | 165 +++++++++++++++++++
.../src/test/resources/setup.script | 30 ++--
.../DeleteByQueryElasticsearch.java | 145 ++++++++++++++++
.../ElasticSearchRestProcessor.java | 91 ++++++++++
.../elasticsearch/JsonQueryElasticsearch.java | 71 +-------
.../org.apache.nifi.processor.Processor | 2 +
.../DeleteByQueryElasticsearchTest.java | 134 +++++++++++++++
.../TestElasticSearchClientService.java | 47 +++++-
15 files changed, 866 insertions(+), 203 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/3a29c1e4/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/DeleteOperationResponse.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/DeleteOperationResponse.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/DeleteOperationResponse.java
new file mode 100644
index 0000000..1fd83d7
--- /dev/null
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/DeleteOperationResponse.java
@@ -0,0 +1,30 @@
+/*
+ * 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;
+
+public class DeleteOperationResponse {
+ private long took;
+
+ public DeleteOperationResponse(long took) {
+ this.took = took;
+ }
+
+ public long getTook() {
+ return took;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/3a29c1e4/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientService.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientService.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientService.java
index 6854c55..188c7bb 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientService.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientService.java
@@ -25,6 +25,8 @@ import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.ssl.SSLContextService;
import java.io.IOException;
+import java.util.List;
+import java.util.Map;
@Tags({"elasticsearch", "client"})
@CapabilityDescription("A controller service for accessing an ElasticSearch client.")
@@ -96,6 +98,66 @@ public interface ElasticSearchClientService extends ControllerService {
.build();
/**
+ * Index a document.
+ *
+ * @param operation A document to index.
+ * @return IndexOperationResponse if successful
+ * @throws IOException thrown when there is an error.
+ */
+ IndexOperationResponse add(IndexOperationRequest operation) throws IOException;
+
+ /**
+ * Index multiple documents.
+ *
+ * @param operations A list of documents to index.
+ * @return IndexOperationResponse if successful.
+ * @throws IOException thrown when there is an error.
+ */
+ IndexOperationResponse add(List<IndexOperationRequest> operations) throws IOException;
+
+ /**
+ * Delete a document by its ID from an index.
+ *
+ * @param index The index to target.
+ * @param type The type to target. Optional.
+ * @param id The document ID to remove from the selected index.
+ * @return A DeleteOperationResponse object if successful.
+ */
+ DeleteOperationResponse deleteById(String index, String type, String id) throws IOException;
+
+
+ /**
+ * Delete multiple documents by ID from an index.
+ * @param index The index to target.
+ * @param type The type to target. Optional.
+ * @param ids A list of document IDs to remove from the selected index.
+ * @return A DeleteOperationResponse object if successful.
+ * @throws IOException thrown when there is an error.
+ */
+ DeleteOperationResponse deleteById(String index, String type, List<String> ids) throws IOException;
+
+ /**
+ * Delete documents by query.
+ *
+ * @param query A valid JSON query to be used for finding documents to delete.
+ * @param index The index to target.
+ * @param type The type to target within the index. Optional.
+ * @return A DeleteOperationResponse object if successful.
+ */
+ DeleteOperationResponse deleteByQuery(String query, String index, String type) throws IOException;
+
+ /**
+ * Get a document by ID.
+ *
+ * @param index The index that holds the document.
+ * @param type The document type. Optional.
+ * @param id The document ID
+ * @return Map if successful, null if not found.
+ * @throws IOException thrown when there is an error.
+ */
+ Map<String, Object> get(String index, String type, String id) throws IOException;
+
+ /**
* Perform a search using the JSON DSL.
*
* @param query A JSON string reprensenting the query.
http://git-wip-us.apache.org/repos/asf/nifi/blob/3a29c1e4/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/IndexOperationRequest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/IndexOperationRequest.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/IndexOperationRequest.java
new file mode 100644
index 0000000..9281adb
--- /dev/null
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/IndexOperationRequest.java
@@ -0,0 +1,50 @@
+/*
+ * 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 java.util.Map;
+
+public class IndexOperationRequest {
+ private String index;
+ private String type;
+ private String id;
+ private Map<String, Object> fields;
+
+ public IndexOperationRequest(String index, String type, String id, Map<String, Object> fields) {
+ this.index = index;
+ this.type = type;
+ this.id = id;
+ this.fields = fields;
+ }
+
+ public String getIndex() {
+ return index;
+ }
+
+ public String getType() {
+ return type;
+ }
+
+ public String getId() {
+ return id;
+ }
+
+ public Map<String, Object> getFields() {
+ return fields;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/3a29c1e4/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/IndexOperationResponse.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/IndexOperationResponse.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/IndexOperationResponse.java
new file mode 100644
index 0000000..a22b7aa
--- /dev/null
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/IndexOperationResponse.java
@@ -0,0 +1,36 @@
+/*
+ * 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;
+
+public class IndexOperationResponse {
+ private long took;
+ private long ingestTook;
+
+ public IndexOperationResponse(long took, long ingestTook) {
+ this.took = took;
+ this.ingestTook = ingestTook;
+ }
+
+ public long getTook() {
+ return took;
+ }
+
+ public long getIngestTook() {
+ return ingestTook;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/3a29c1e4/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 8f52ab8..db79b05 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
@@ -70,11 +70,6 @@
</dependency>
<dependency>
- <groupId>org.elasticsearch.client</groupId>
- <artifactId>rest</artifactId>
- <version>5.0.1</version>
- </dependency>
- <dependency>
<groupId>com.github.stephenc.findbugs</groupId>
<artifactId>findbugs-annotations</artifactId>
<version>1.3.9-1</version>
@@ -120,6 +115,18 @@
<version>1.7.0-SNAPSHOT</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.elasticsearch</groupId>
+ <artifactId>elasticsearch</artifactId>
+ <version>5.6.8</version>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.elasticsearch.client</groupId>
+ <artifactId>elasticsearch-rest-high-level-client</artifactId>
+ <version>5.6.8</version>
+ <scope>compile</scope>
+ </dependency>
</dependencies>
<profiles>
@@ -137,7 +144,6 @@
<httpPort>9400</httpPort>
<version>5.6.2</version>
<timeout>90</timeout>
- <pathInitScript>${project.basedir}/src/test/resources/setup.script</pathInitScript>
</configuration>
<executions>
<execution>
http://git-wip-us.apache.org/repos/asf/nifi/blob/3a29c1e4/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
index 9896a19..111490b 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
@@ -19,6 +19,7 @@ package org.apache.nifi.elasticsearch;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.io.IOUtils;
+import org.apache.http.Header;
import org.apache.http.HttpEntity;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
@@ -34,9 +35,17 @@ import org.apache.nifi.controller.AbstractControllerService;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.ssl.SSLContextService;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.get.GetRequest;
+import org.elasticsearch.action.get.GetResponse;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
+import org.elasticsearch.client.RestHighLevelClient;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
@@ -55,6 +64,7 @@ import java.security.SecureRandom;
import java.security.UnrecoverableKeyException;
import java.security.cert.CertificateException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -66,6 +76,7 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
static final private List<PropertyDescriptor> properties;
private RestClient client;
+ private RestHighLevelClient highLevelClient;
private String url;
private Charset charset;
@@ -182,18 +193,19 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
.setMaxRetryTimeoutMillis(retryTimeout);
this.client = builder.build();
+ this.highLevelClient = new RestHighLevelClient(client);
}
- private Response runQuery(String query, String index, String type) throws IOException {
+ private Response runQuery(String endpoint, String query, String index, String type) throws IOException {
StringBuilder sb = new StringBuilder()
- .append("/")
- .append(index);
+ .append("/")
+ .append(index);
if (type != null && !type.equals("")) {
sb.append("/")
- .append(type);
+ .append(type);
}
- sb.append("/_search");
+ sb.append(String.format("/%s", endpoint));
HttpEntity queryEntity = new NStringEntity(query, ContentType.APPLICATION_JSON);
@@ -216,8 +228,67 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
}
@Override
+ public IndexOperationResponse add(IndexOperationRequest operation) throws IOException {
+ return add(Arrays.asList(operation));
+ }
+
+ @Override
+ public IndexOperationResponse add(List<IndexOperationRequest> operations) throws IOException {
+ BulkRequest bulkRequest = new BulkRequest();
+ for (int index = 0; index < operations.size(); index++) {
+ IndexOperationRequest or = operations.get(index);
+ IndexRequest indexRequest = new IndexRequest(or.getIndex(), or.getType(), or.getId())
+ .source(or.getFields());
+ bulkRequest.add(indexRequest);
+ }
+
+ bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+
+ BulkResponse response = highLevelClient.bulk(bulkRequest);
+ IndexOperationResponse retVal = new IndexOperationResponse(response.getTookInMillis(), response.getIngestTookInMillis());
+
+ return retVal;
+ }
+
+ @Override
+ public DeleteOperationResponse deleteById(String index, String type, String id) throws IOException {
+ return deleteById(index, type, Arrays.asList(id));
+ }
+
+ @Override
+ public DeleteOperationResponse deleteById(String index, String type, List<String> ids) throws IOException {
+ BulkRequest bulk = new BulkRequest();
+ for (int idx = 0; idx < ids.size(); idx++) {
+ DeleteRequest request = new DeleteRequest(index, type, ids.get(idx));
+ bulk.add(request);
+ }
+ BulkResponse response = highLevelClient.bulk(bulk);
+
+ DeleteOperationResponse dor = new DeleteOperationResponse(response.getTookInMillis());
+
+ return dor;
+ }
+
+ @Override
+ public DeleteOperationResponse deleteByQuery(String query, String index, String type) throws IOException {
+ long start = System.currentTimeMillis();
+ Response response = runQuery("_delete_by_query", query, index, type);
+ long end = System.currentTimeMillis();
+ Map<String, Object> parsed = parseResponse(response);
+
+ return new DeleteOperationResponse(end - start);
+ }
+
+ @Override
+ public Map<String, Object> get(String index, String type, String id) throws IOException {
+ GetRequest get = new GetRequest(index, type, id);
+ GetResponse resp = highLevelClient.get(get, new Header[]{});
+ return resp.getSource();
+ }
+
+ @Override
public SearchResponse search(String query, String index, String type) throws IOException {
- Response response = runQuery(query, index, type);
+ Response response = runQuery("_search", query, index, type);
Map<String, Object> parsed = parseResponse(response);
int took = (Integer)parsed.get("took");
http://git-wip-us.apache.org/repos/asf/nifi/blob/3a29c1e4/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/ElasticSearch5ClientService_IT.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/ElasticSearch5ClientService_IT.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/ElasticSearch5ClientService_IT.java
deleted file mode 100644
index 7ab618d..0000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/ElasticSearch5ClientService_IT.java
+++ /dev/null
@@ -1,105 +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.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 java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-public class ElasticSearch5ClientService_IT {
-
- private TestRunner runner;
- private ElasticSearchClientServiceImpl service;
-
- @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;
- }
- }
-
- @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\",\n" +
- "\t\t\t\t\"size\": 5\n" +
- "\t\t\t}\n" +
- "\t\t}\n" +
- "\t}\n" +
- "}";
- SearchResponse response = service.search(query, "messages", "message");
- Assert.assertNotNull("Response was null", response);
-
- Assert.assertEquals("Wrong count", response.getNumberOfHits(), 15);
- 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);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/nifi/blob/3a29c1e4/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
new file mode 100644
index 0000000..687faf0
--- /dev/null
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/ElasticSearchClientService_IT.java
@@ -0,0 +1,165 @@
+/*
+ * 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/3a29c1e4/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 69671b6..8cf4c97 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
@@ -15,18 +15,18 @@
#create mapping
PUT:messages/:{ "mappings":{"message":{ "properties":{ "msg":{"type":"keyword"}}}}}
#add document
-PUT:messages/message/1:{ "msg ":"one" }
-PUT:messages/message/2:{ "msg ":"two" }
-PUT:messages/message/3:{ "msg ":"two" }
-PUT:messages/message/4:{ "msg ":"three" }
-PUT:messages/message/5:{ "msg ":"three" }
-PUT:messages/message/6:{ "msg ":"three" }
-PUT:messages/message/7:{ "msg ":"four" }
-PUT:messages/message/8:{ "msg ":"four" }
-PUT:messages/message/9:{ "msg ":"four" }
-PUT:messages/message/10:{ "msg ":"four" }
-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/1:{ "msg":"one" }
+PUT:messages/message/2:{ "msg":"two" }
+PUT:messages/message/3:{ "msg":"two" }
+PUT:messages/message/4:{ "msg":"three" }
+PUT:messages/message/5:{ "msg":"three" }
+PUT:messages/message/6:{ "msg":"three" }
+PUT:messages/message/7:{ "msg":"four" }
+PUT:messages/message/8:{ "msg":"four" }
+PUT:messages/message/9:{ "msg":"four" }
+PUT:messages/message/10:{ "msg":"four" }
+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
http://git-wip-us.apache.org/repos/asf/nifi/blob/3a29c1e4/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearch.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearch.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearch.java
new file mode 100644
index 0000000..609d8ed
--- /dev/null
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearch.java
@@ -0,0 +1,145 @@
+/*
+ * 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 org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+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.PropertyDescriptor;
+import org.apache.nifi.elasticsearch.DeleteOperationResponse;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.util.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_ALLOWED)
+@CapabilityDescription("Delete from an ElasticSearch index using a query. The query can be loaded from a flowfile body " +
+ "or from the Query parameter.")
+@Tags({ "elastic", "elasticsearch", "delete", "query"})
+@WritesAttributes({
+ @WritesAttribute(attribute = "elasticsearch.delete.took", description = "The amount of time that it took to complete the delete operation in ms."),
+ @WritesAttribute(attribute = "elasticsearch.delete.error", description = "The error message provided by ElasticSearch if there is an error running the delete.")
+})
+public class DeleteByQueryElasticsearch extends AbstractProcessor implements ElasticSearchRestProcessor {
+ public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure")
+ .description("If the delete by query fails, and a flowfile was read, it will be sent to this relationship.").build();
+
+ public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success")
+ .description("If the delete by query succeeds, and a flowfile was read, it will be sent to this relationship.")
+ .build();
+
+
+ static final String TOOK_ATTRIBUTE = "elasticsearch.delete.took";
+ static final String ERROR_ATTRIBUTE = "elasticsearch.delete.error";
+
+ private static final Set<Relationship> relationships;
+ private static final List<PropertyDescriptor> propertyDescriptors;
+
+ private volatile ElasticSearchClientService clientService;
+
+ static {
+ final Set<Relationship> _rels = new HashSet<>();
+ _rels.add(REL_SUCCESS);
+ _rels.add(REL_FAILURE);
+ relationships = Collections.unmodifiableSet(_rels);
+
+ final List<PropertyDescriptor> descriptors = new ArrayList<>();
+ descriptors.add(QUERY);
+ descriptors.add(QUERY_ATTRIBUTE);
+ descriptors.add(INDEX);
+ descriptors.add(TYPE);
+ descriptors.add(CLIENT_SERVICE);
+
+ propertyDescriptors = Collections.unmodifiableList(descriptors);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return propertyDescriptors;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ clientService = context.getProperty(CLIENT_SERVICE).asControllerService(ElasticSearchClientService.class);
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ FlowFile input = null;
+ if (context.hasIncomingConnection()) {
+ input = session.get();
+
+ if (input == null && context.hasNonLoopConnection()) {
+ return;
+ }
+ }
+
+ try {
+ final String query = getQuery(input, context, session);
+ final String index = context.getProperty(INDEX).evaluateAttributeExpressions(input).getValue();
+ final String type = context.getProperty(TYPE).isSet()
+ ? context.getProperty(TYPE).evaluateAttributeExpressions(input).getValue()
+ : null;
+ final String queryAttr = context.getProperty(QUERY_ATTRIBUTE).isSet()
+ ? context.getProperty(QUERY_ATTRIBUTE).evaluateAttributeExpressions(input).getValue()
+ : null;
+ DeleteOperationResponse dor = clientService.deleteByQuery(query, index, type);
+
+ if (input == null) {
+ input = session.create();
+ }
+
+ Map<String, String> attrs = new HashMap<>();
+ attrs.put(TOOK_ATTRIBUTE, String.valueOf(dor.getTook()));
+ if (!StringUtils.isBlank(queryAttr)) {
+ attrs.put(queryAttr, query);
+ }
+
+ input = session.putAllAttributes(input, attrs);
+
+ session.transfer(input, REL_SUCCESS);
+ } catch (Exception e) {
+ if (input != null) {
+ input = session.putAttribute(input, ERROR_ATTRIBUTE, e.getMessage());
+ session.transfer(input, REL_FAILURE);
+ }
+ getLogger().error("Error running delete by query: ", e);
+ context.yield();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/3a29c1e4/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/ElasticSearchRestProcessor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/ElasticSearchRestProcessor.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/ElasticSearchRestProcessor.java
new file mode 100644
index 0000000..e94a1cc
--- /dev/null
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/ElasticSearchRestProcessor.java
@@ -0,0 +1,91 @@
+/*
+ * 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 org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.Validator;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+
+public interface ElasticSearchRestProcessor {
+ PropertyDescriptor INDEX = new PropertyDescriptor.Builder()
+ .name("el-rest-fetch-index")
+ .displayName("Index")
+ .description("The name of the index to use.")
+ .required(true)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ PropertyDescriptor TYPE = new PropertyDescriptor.Builder()
+ .name("el-rest-type")
+ .displayName("Type")
+ .description("The type of this document (used by Elasticsearch for indexing and searching)")
+ .required(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ PropertyDescriptor QUERY = new PropertyDescriptor.Builder()
+ .name("el-rest-query")
+ .displayName("Query")
+ .description("A query in JSON syntax, not Lucene syntax. Ex: {\"query\":{\"match\":{\"somefield\":\"somevalue\"}}}. " +
+ "If this parameter is not set, the query will be read from the flowfile content.")
+ .required(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ PropertyDescriptor QUERY_ATTRIBUTE = new PropertyDescriptor.Builder()
+ .name("el-query-attribute")
+ .displayName("Query Attribute")
+ .description("If set, the executed query will be set on each result flowfile in the specified attribute.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(Validator.VALID)
+ .required(false)
+ .build();
+
+ 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();
+
+ default String getQuery(FlowFile input, ProcessContext context, ProcessSession session) throws IOException {
+ String retVal = null;
+ if (context.getProperty(QUERY).isSet()) {
+ retVal = context.getProperty(QUERY).evaluateAttributeExpressions(input).getValue();
+ } else if (input != null) {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ session.exportTo(input, out);
+ out.close();
+
+ retVal = new String(out.toByteArray());
+ }
+
+ return retVal;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/3a29c1e4/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/JsonQueryElasticsearch.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/JsonQueryElasticsearch.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/JsonQueryElasticsearch.java
index 64be962..40d6118 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/JsonQueryElasticsearch.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/JsonQueryElasticsearch.java
@@ -27,7 +27,6 @@ import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.components.AllowableValue;
import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.components.Validator;
import org.apache.nifi.elasticsearch.ElasticSearchClientService;
import org.apache.nifi.elasticsearch.SearchResponse;
import org.apache.nifi.expression.ExpressionLanguageScope;
@@ -38,10 +37,8 @@ import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.util.StringUtils;
-import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
@@ -62,7 +59,7 @@ import java.util.Set;
"ElasticSearch JSON DSL. It does not automatically paginate queries for the user. If an incoming relationship is added to this " +
"processor, it will use the flowfile's content for the query. Care should be taken on the size of the query because the entire response " +
"from ElasticSearch will be loaded into memory all at once and converted into the resulting flowfiles.")
-public class JsonQueryElasticsearch extends AbstractProcessor {
+public class JsonQueryElasticsearch extends AbstractProcessor implements ElasticSearchRestProcessor {
public static final Relationship REL_ORIGINAL = new Relationship.Builder().name("original")
.description("All original flowfiles that don't cause an error to occur go to this relationship. " +
"This applies even if you select the \"split up hits\" option to send individual hits to the " +
@@ -78,49 +75,6 @@ public class JsonQueryElasticsearch extends AbstractProcessor {
public static final Relationship REL_AGGREGATIONS = new Relationship.Builder().name("aggregations")
.description("Aggregations are routed to this relationship.")
.build();
-
- public static final PropertyDescriptor INDEX = new PropertyDescriptor.Builder()
- .name("el-rest-fetch-index")
- .displayName("Index")
- .description("The name of the index to read from")
- .required(true)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor TYPE = new PropertyDescriptor.Builder()
- .name("el-rest-type")
- .displayName("Type")
- .description("The type of this document (used by Elasticsearch for indexing and searching)")
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor QUERY = new PropertyDescriptor.Builder()
- .name("el-rest-query")
- .displayName("Query")
- .description("A query in JSON syntax, not Lucene syntax. Ex: " +
- "{\n" +
- "\t\"query\": {\n" +
- "\t\t\"match\": {\n" +
- "\t\t\t\"name\": \"John Smith\"\n" +
- "\t\t}\n" +
- "\t}\n" +
- "}")
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
- public static final PropertyDescriptor QUERY_ATTRIBUTE = new PropertyDescriptor.Builder()
- .name("el-query-attribute")
- .displayName("Query Attribute")
- .description("If set, the executed query will be set on each result flowfile in the specified attribute.")
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(Validator.VALID)
- .required(false)
- .build();
-
public static final AllowableValue SPLIT_UP_YES = new AllowableValue(
"splitUp-yes",
"Yes",
@@ -151,14 +105,6 @@ public class JsonQueryElasticsearch extends AbstractProcessor {
.expressionLanguageSupported(ExpressionLanguageScope.NONE)
.build();
- 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();
-
private static final Set<Relationship> relationships;
private static final List<PropertyDescriptor> propertyDescriptors;
@@ -207,21 +153,6 @@ public class JsonQueryElasticsearch extends AbstractProcessor {
private final ObjectMapper mapper = new ObjectMapper();
- private String getQuery(FlowFile input, ProcessContext context, ProcessSession session) throws IOException {
- String retVal = null;
- if (context.getProperty(QUERY).isSet()) {
- retVal = context.getProperty(QUERY).evaluateAttributeExpressions(input).getValue();
- } else if (input != null) {
- ByteArrayOutputStream out = new ByteArrayOutputStream();
- session.exportTo(input, out);
- out.close();
-
- retVal = new String(out.toByteArray());
- }
-
- return retVal;
- }
-
@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
FlowFile input = null;
http://git-wip-us.apache.org/repos/asf/nifi/blob/3a29c1e4/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
index d57ff36..ac8c915 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
@@ -12,4 +12,6 @@
# 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.processors.elasticsearch.DeleteByQueryElasticsearch
org.apache.nifi.processors.elasticsearch.JsonQueryElasticsearch
http://git-wip-us.apache.org/repos/asf/nifi/blob/3a29c1e4/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearchTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearchTest.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearchTest.java
new file mode 100644
index 0000000..ade1102
--- /dev/null
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearchTest.java
@@ -0,0 +1,134 @@
+/*
+ * 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 org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class DeleteByQueryElasticsearchTest {
+ private static final String INDEX = "test_idx";
+ private static final String TYPE = "test_type";
+ private static final String QUERY_ATTR = "es.delete.query";
+ private static final String CLIENT_NAME = "clientService";
+
+ private TestElasticSearchClientService client;
+
+ private void initClient(TestRunner runner) throws Exception {
+ client = new TestElasticSearchClientService(true);
+ runner.addControllerService(CLIENT_NAME, client);
+ runner.enableControllerService(client);
+ runner.setProperty(DeleteByQueryElasticsearch.CLIENT_SERVICE, CLIENT_NAME);
+ }
+
+ private void postTest(TestRunner runner, String queryParam) {
+ runner.assertTransferCount(DeleteByQueryElasticsearch.REL_FAILURE, 0);
+ runner.assertTransferCount(DeleteByQueryElasticsearch.REL_SUCCESS, 1);
+
+ List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(DeleteByQueryElasticsearch.REL_SUCCESS);
+ String attr = flowFiles.get(0).getAttribute(DeleteByQueryElasticsearch.TOOK_ATTRIBUTE);
+ String query = flowFiles.get(0).getAttribute(QUERY_ATTR);
+ Assert.assertNotNull(attr);
+ Assert.assertEquals(attr, "100");
+ Assert.assertNotNull(query);
+ Assert.assertEquals(queryParam, query);
+ }
+
+ @Test
+ public void testWithFlowfileInput() throws Exception {
+ String query = "{ \"query\": { \"match_all\": {} }}";
+ TestRunner runner = TestRunners.newTestRunner(DeleteByQueryElasticsearch.class);
+ runner.setProperty(DeleteByQueryElasticsearch.INDEX, INDEX);
+ runner.setProperty(DeleteByQueryElasticsearch.TYPE, TYPE);
+ runner.setProperty(DeleteByQueryElasticsearch.QUERY_ATTRIBUTE, QUERY_ATTR);
+ initClient(runner);
+ runner.assertValid();
+ runner.enqueue(query);
+ runner.run();
+
+ postTest(runner, query);
+ }
+
+ @Test
+ public void testWithQuery() throws Exception {
+ String query = "{\n" +
+ "\t\"query\": {\n" +
+ "\t\t\"match\": {\n" +
+ "\t\t\t\"${field.name}.keyword\": \"test\"\n" +
+ "\t\t}\n" +
+ "\t}\n" +
+ "}";
+ Map<String, String> attrs = new HashMap<String, String>(){{
+ put("field.name", "test_field");
+ }};
+ TestRunner runner = TestRunners.newTestRunner(DeleteByQueryElasticsearch.class);
+ runner.setProperty(DeleteByQueryElasticsearch.QUERY, query);
+ runner.setProperty(DeleteByQueryElasticsearch.INDEX, INDEX);
+ runner.setProperty(DeleteByQueryElasticsearch.TYPE, TYPE);
+ runner.setProperty(DeleteByQueryElasticsearch.QUERY_ATTRIBUTE, QUERY_ATTR);
+ initClient(runner);
+ runner.assertValid();
+ runner.enqueue("", attrs);
+ runner.run();
+
+ postTest(runner, query.replace("${field.name}", "test_field"));
+
+ runner.clearTransferState();
+
+ query = "{\n" +
+ "\t\"query\": {\n" +
+ "\t\t\"match\": {\n" +
+ "\t\t\t\"test_field.keyword\": \"test\"\n" +
+ "\t\t}\n" +
+ "\t}\n" +
+ "}";
+ runner.setProperty(DeleteByQueryElasticsearch.QUERY, query);
+ runner.setIncomingConnection(false);
+ runner.assertValid();
+ runner.run();
+ postTest(runner, query);
+ }
+
+ @Test
+ public void testErrorAttribute() throws Exception {
+ String query = "{ \"query\": { \"match_all\": {} }}";
+ TestRunner runner = TestRunners.newTestRunner(DeleteByQueryElasticsearch.class);
+ runner.setProperty(DeleteByQueryElasticsearch.QUERY, query);
+ runner.setProperty(DeleteByQueryElasticsearch.INDEX, INDEX);
+ runner.setProperty(DeleteByQueryElasticsearch.TYPE, TYPE);
+ runner.setProperty(DeleteByQueryElasticsearch.QUERY_ATTRIBUTE, QUERY_ATTR);
+ initClient(runner);
+ client.setThrowErrorInDelete(true);
+ runner.assertValid();
+ runner.enqueue("");
+ runner.run();
+
+ runner.assertTransferCount(DeleteByQueryElasticsearch.REL_SUCCESS, 0);
+ runner.assertTransferCount(DeleteByQueryElasticsearch.REL_FAILURE, 1);
+
+ MockFlowFile mockFlowFile = runner.getFlowFilesForRelationship(DeleteByQueryElasticsearch.REL_FAILURE).get(0);
+ String attr = mockFlowFile.getAttribute(DeleteByQueryElasticsearch.ERROR_ATTRIBUTE);
+ Assert.assertNotNull(attr);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/3a29c1e4/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestElasticSearchClientService.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestElasticSearchClientService.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestElasticSearchClientService.java
index 445a093..b1d4220 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestElasticSearchClientService.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestElasticSearchClientService.java
@@ -19,25 +19,66 @@ package org.apache.nifi.processors.elasticsearch;
import com.fasterxml.jackson.databind.ObjectMapper;
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;
import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class TestElasticSearchClientService extends AbstractControllerService implements ElasticSearchClientService {
private boolean returnAggs;
private boolean throwErrorInSearch;
+ private boolean throwErrorInDelete;
public TestElasticSearchClientService(boolean returnAggs) {
this.returnAggs = returnAggs;
}
@Override
+ public IndexOperationResponse add(IndexOperationRequest operation) throws IOException {
+ return add(Arrays.asList(operation));
+ }
+
+ @Override
+ public IndexOperationResponse add(List<IndexOperationRequest> operations) throws IOException {
+ return new IndexOperationResponse(100L, 100L);
+ }
+
+ @Override
+ public DeleteOperationResponse deleteById(String index, String type, String id) throws IOException {
+ return deleteById(index, type, Arrays.asList(id));
+ }
+
+ @Override
+ public DeleteOperationResponse deleteById(String index, String type, List<String> ids) throws IOException {
+ if (throwErrorInDelete) {
+ throw new IOException("Simulated IOException");
+ }
+ return new DeleteOperationResponse(100L);
+ }
+
+ @Override
+ public DeleteOperationResponse deleteByQuery(String query, String index, String type) throws IOException {
+ return deleteById(index, type, Arrays.asList("1"));
+ }
+
+ @Override
+ public Map<String, Object> get(String index, String type, String id) throws IOException {
+ return new HashMap<String, Object>(){{
+ put("msg", "one");
+ }};
+ }
+
+ @Override
public SearchResponse search(String query, String index, String type) throws IOException {
if (throwErrorInSearch) {
- throw new IOException();
+ throw new IOException("Simulated IOException");
}
ObjectMapper mapper = new ObjectMapper();
@@ -203,4 +244,8 @@ public class TestElasticSearchClientService extends AbstractControllerService im
public void setThrowErrorInSearch(boolean throwErrorInSearch) {
this.throwErrorInSearch = throwErrorInSearch;
}
+
+ public void setThrowErrorInDelete(boolean throwErrorInDelete) {
+ this.throwErrorInDelete = throwErrorInDelete;
+ }
}