Trace method(clearScrollAsync) for elasticsearch-6.x-plugin and polishing elasticsearch-6.x-scenario. (#6031)

* Polish elasticsearch-6.x-scenario.
2020-12-19 20:58:07 +08:00 · 2020-12-19 20:58:07 +08:00 · de3ccecd22
parent 5e9582eea9
commit de3ccecd22
3 changed files with 496 additions and 20 deletions
--- a/apm-sniffer/apm-sdk-plugin/elasticsearch-6.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/elasticsearch/v6/define/RestHighLevelClientInstrumentation.java
+++ b/apm-sniffer/apm-sdk-plugin/elasticsearch-6.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/elasticsearch/v6/define/RestHighLevelClientInstrumentation.java
@ -18,11 +18,6 @@

 package org.apache.skywalking.apm.plugin.elasticsearch.v6.define;

-import static net.bytebuddy.matcher.ElementMatchers.named;
-import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
-import static org.apache.skywalking.apm.agent.core.plugin.bytebuddy.ArgumentTypeNameMatch.takesArgumentWithType;
-import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
-
 import net.bytebuddy.description.method.MethodDescription;
 import net.bytebuddy.matcher.ElementMatcher;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
@ -32,6 +27,11 @@ import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassEnha
 import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
 import org.apache.skywalking.apm.plugin.elasticsearch.v6.interceptor.Constants;

+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+import static org.apache.skywalking.apm.agent.core.plugin.bytebuddy.ArgumentTypeNameMatch.takesArgumentWithType;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
+
 /**
 * {@link RestHighLevelClientInstrumentation} enhance the constructor method without argument in
 * <code>org.elasticsearch.client.RestHighLevelClient</code> by <code>org.apache.skywalking.apm.plugin.elasticsearch.v6.interceptor.RestHighLevelClientConInterceptor</code>
@ -221,7 +221,7 @@ public class RestHighLevelClientInstrumentation extends ClassEnhancePluginDefine
            new InstanceMethodsInterceptPoint() {
                @Override
                public ElementMatcher<MethodDescription> getMethodsMatcher() {
-                    return named("clearScroll");
+                    return named("clearScroll").or(named("clearScrollAsync"));
                }

                @Override
--- a/test/plugin/scenarios/elasticsearch-6.x-scenario/config/expectedData.yaml
+++ b/test/plugin/scenarios/elasticsearch-6.x-scenario/config/expectedData.yaml
@ -82,8 +82,7 @@ segmentItems:
      - {key: db.type, value: Elasticsearch}
      - {key: db.instance, value: not null}
      - {key: db.statement, value: not null}
-      skipAnalysis: 'false'
-    - operationName: Elasticsearch/UpdateRequest
+    - operationName: Elasticsearch/SearchRequest
      operationId: 0
      parentSpanId: 0
      spanId: 5
@ -94,15 +93,229 @@ segmentItems:
      isError: false
      spanType: Exit
      peer: not null
+      tags:
+        - {key: db.type, value: Elasticsearch}
+        - {key: db.instance, value: not null}
+        - {key: db.statement, value: not null}
+    - operationName: Elasticsearch/SearchScrollRequest
+      operationId: 0
+      parentSpanId: 0
+      spanId: 6
+      spanLayer: Database
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 77
+      isError: false
+      spanType: Exit
+      peer: not null
+      tags:
+        - {key: db.type, value: Elasticsearch}
+        - {key: db.statement, value: not null}
+      skipAnalysis: 'false'
+    - operationName: Elasticsearch/SearchRequest
+      operationId: 0
+      parentSpanId: 0
+      spanId: 7
+      spanLayer: Database
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 77
+      isError: false
+      spanType: Exit
+      peer: not null
+      tags:
+        - {key: db.type, value: Elasticsearch}
+        - {key: db.instance, value: not null}
+        - {key: db.statement, value: not null}
+    - operationName: Elasticsearch/SearchScrollRequest
+      operationId: 0
+      parentSpanId: 0
+      spanId: 8
+      spanLayer: Database
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 77
+      isError: false
+      spanType: Exit
+      peer: not null
+      tags:
+        - {key: db.type, value: Elasticsearch}
+        - {key: db.statement, value: not null}
+      skipAnalysis: 'false'
+    - operationName: Elasticsearch/SearchRequest
+      operationId: 0
+      parentSpanId: 0
+      spanId: 9
+      spanLayer: Database
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 77
+      isError: false
+      spanType: Exit
+      peer: not null
+      tags:
+        - {key: db.type, value: Elasticsearch}
+        - {key: db.instance, value: not null}
+        - {key: db.statement, value: not null}
+    - operationName: Elasticsearch/ClearScrollRequest
+      operationId: 0
+      parentSpanId: 0
+      spanId: 10
+      spanLayer: Database
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 77
+      isError: false
+      spanType: Exit
+      peer: not null
+      tags:
+        - {key: db.type, value: Elasticsearch}
+        - {key: db.statement, value: not null}
+      skipAnalysis: 'false'
+    - operationName: Elasticsearch/SearchRequest
+      operationId: 0
+      parentSpanId: 0
+      spanId: 11
+      spanLayer: Database
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 77
+      isError: false
+      spanType: Exit
+      peer: not null
+      tags:
+        - {key: db.type, value: Elasticsearch}
+        - {key: db.instance, value: not null}
+        - {key: db.statement, value: not null}
+    - operationName: Elasticsearch/ClearScrollRequest
+      operationId: 0
+      parentSpanId: 0
+      spanId: 12
+      spanLayer: Database
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 77
+      isError: false
+      spanType: Exit
+      peer: not null
+      tags:
+        - {key: db.type, value: Elasticsearch}
+        - {key: db.statement, value: not null}
+      skipAnalysis: 'false'
+    - operationName: Elasticsearch/SearchTemplateRequest
+      operationId: 0
+      parentSpanId: 0
+      spanId: 13
+      spanLayer: Database
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 77
+      isError: false
+      spanType: Exit
+      peer: not null
+      tags:
+        - {key: db.type, value: Elasticsearch}
+        - {key: db.instance, value: not null}
+        - {key: db.statement, value: not null}
+    - operationName: Elasticsearch/SearchTemplateRequest
+      operationId: 0
+      parentSpanId: 0
+      spanId: 14
+      spanLayer: Database
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 77
+      isError: false
+      spanType: Exit
+      peer: not null
+      tags:
+        - {key: db.type, value: Elasticsearch}
+        - {key: db.instance, value: not null}
+        - {key: db.statement, value: not null}
+    - operationName: Elasticsearch/UpdateRequest
+      operationId: 0
+      parentSpanId: 0
+      spanId: 15
+      spanLayer: Database
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 77
+      isError: false
+      spanType: Exit
+      peer: not null
      tags:
      - {key: db.type, value: Elasticsearch}
      - {key: db.instance, value: not null}
      - {key: db.statement, value: not null}
      skipAnalysis: 'false'
+    - operationName: Elasticsearch/AnalyzeRequest
+      operationId: 0
+      parentSpanId: 0
+      spanId: 16
+      spanLayer: Database
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 77
+      isError: false
+      spanType: Exit
+      peer: not null
+      tags:
+        - {key: db.type, value: Elasticsearch}
+        - {key: analyzer, value: ''}
+        - {key: db.statement, value: not null}
+      skipAnalysis: 'false'
+    - operationName: Elasticsearch/AnalyzeRequest
+      operationId: 0
+      parentSpanId: 0
+      spanId: 17
+      spanLayer: Database
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 77
+      isError: false
+      spanType: Exit
+      peer: not null
+      tags:
+        - {key: db.type, value: Elasticsearch}
+        - {key: analyzer, value: ''}
+        - {key: db.statement, value: not null}
+      skipAnalysis: 'false'
+    - operationName: Elasticsearch/DeleteByQueryRequest
+      operationId: 0
+      parentSpanId: 0
+      spanId: 18
+      spanLayer: Database
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 77
+      isError: false
+      spanType: Exit
+      peer: not null
+      tags:
+        - {key: db.type, value: Elasticsearch}
+        - {key: db.instance, value: not null}
+        - {key: db.statement, value: not null}
+      skipAnalysis: 'false'
+    - operationName: Elasticsearch/DeleteByQueryRequest
+      operationId: 0
+      parentSpanId: 0
+      spanId: 19
+      spanLayer: Database
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 77
+      isError: false
+      spanType: Exit
+      peer: not null
+      tags:
+        - {key: db.type, value: Elasticsearch}
+        - {key: db.instance, value: not null}
+        - {key: db.statement, value: not null}
+      skipAnalysis: 'false'
    - operationName: Elasticsearch/DeleteRequest
      operationId: 0
      parentSpanId: 0
-      spanId: 6
+      spanId: 20
      spanLayer: Database
      startTime: nq 0
      endTime: nq 0
@ -117,7 +330,7 @@ segmentItems:
    - operationName: Elasticsearch/IndexRequest
      operationId: 0
      parentSpanId: 0
-      spanId: 7
+      spanId: 21
      spanLayer: Database
      startTime: nq 0
      endTime: nq 0
@ -136,7 +349,7 @@ segmentItems:
    - operationName: Elasticsearch/actionGet
      operationId: 0
      parentSpanId: 0
-      spanId: 8
+      spanId: 22
      startTime: nq 0
      endTime: nq 0
      componentId: 48
@ -149,7 +362,7 @@ segmentItems:
    - operationName: Elasticsearch/GetRequest
      operationId: 0
      parentSpanId: 0
-      spanId: 9
+      spanId: 23
      spanLayer: Database
      startTime: nq 0
      endTime: nq 0
@ -168,7 +381,7 @@ segmentItems:
    - operationName: Elasticsearch/SearchRequest
      operationId: 0
      parentSpanId: 0
-      spanId: 10
+      spanId: 24
      spanLayer: Database
      startTime: nq 0
      endTime: nq 0
@ -187,7 +400,7 @@ segmentItems:
    - operationName: Elasticsearch/UpdateRequest
      operationId: 0
      parentSpanId: 0
-      spanId: 11
+      spanId: 25
      spanLayer: Database
      startTime: nq 0
      endTime: nq 0
@ -206,7 +419,7 @@ segmentItems:
    - operationName: Elasticsearch/DeleteRequest
      operationId: 0
      parentSpanId: 0
-      spanId: 12
+      spanId: 26
      spanLayer: Database
      startTime: nq 0
      endTime: nq 0
@ -225,7 +438,7 @@ segmentItems:
    - operationName: Elasticsearch/DeleteIndexRequest
      operationId: 0
      parentSpanId: 0
-      spanId: 13
+      spanId: 27
      spanLayer: Database
      startTime: nq 0
      endTime: nq 0
--- a/test/plugin/scenarios/elasticsearch-6.x-scenario/src/main/java/org/apache/skywalking/apm/testcase/elasticsearch/RestHighLevelClientCase.java
+++ b/test/plugin/scenarios/elasticsearch-6.x-scenario/src/main/java/org/apache/skywalking/apm/testcase/elasticsearch/RestHighLevelClientCase.java
@ -18,18 +18,29 @@

 package org.apache.skywalking.apm.testcase.elasticsearch;

+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
 import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
+import org.elasticsearch.action.admin.indices.analyze.AnalyzeRequest;
+import org.elasticsearch.action.admin.indices.analyze.AnalyzeResponse;
 import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
 import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
 import org.elasticsearch.action.get.GetRequest;
 import org.elasticsearch.action.get.GetResponse;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.search.ClearScrollRequest;
+import org.elasticsearch.action.search.ClearScrollResponse;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.search.SearchScrollRequest;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.action.update.UpdateRequest;
 import org.elasticsearch.action.update.UpdateResponse;
@ -43,16 +54,16 @@ import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.index.reindex.BulkByScrollResponse;
+import org.elasticsearch.index.reindex.DeleteByQueryRequest;
 import org.elasticsearch.script.Script;
 import org.elasticsearch.script.ScriptType;
+import org.elasticsearch.script.mustache.SearchTemplateRequest;
+import org.elasticsearch.script.mustache.SearchTemplateResponse;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;

-import java.io.IOException;
-import java.util.Map;
-import java.util.UUID;
-
 import static java.util.Collections.singletonMap;

@Component
@ -91,8 +102,28 @@ public class RestHighLevelClientCase {
            get(client, indexName);
            // search
            search(client, indexName);
+            // scroll
+            scroll(client, indexName);
+            // scrollAsync
+            scrollAsync(client, indexName);
+            // clearScroll
+            clearScroll(client, indexName);
+            // clearScrollAsync
+            clearScrollAsync(client, indexName);
+            // searchTemplate
+            searchTemplate(client, indexName);
+            // searchTemplateAsync
+            searchTemplateAsync(client, indexName);
            // update
            update(client, indexName);
+            // analyze
+            analyze(client, indexName);
+            // analyzeAsync
+            analyzeAsync(client, indexName);
+            // deleteByQuery
+            deleteByQuery(client, indexName);
+            // deleteByQueryAsync
+            deleteByQueryAsync(client, indexName);
            // delete
            delete(client, indexName);
        } finally {
@ -180,6 +211,37 @@ public class RestHighLevelClientCase {
        }
    }

+    private void analyze(RestHighLevelClient client, String indexName) throws IOException {
+        AnalyzeRequest analyzeRequest = new AnalyzeRequest(indexName).text("SkyWalking");
+        AnalyzeResponse analyzeResponse = client.indices().analyze(analyzeRequest, RequestOptions.DEFAULT);
+        if (null == analyzeResponse.getTokens() || analyzeResponse.getTokens().size() < 1) {
+            String message = "elasticsearch analyze index fail.";
+            LOGGER.error(message);
+            throw new RuntimeException(message);
+        }
+    }
+
+    private void analyzeAsync(RestHighLevelClient client, String indexName) throws IOException {
+        ActionListener<AnalyzeResponse> listener = new ActionListener<AnalyzeResponse>() {
+            @Override
+            public void onResponse(final AnalyzeResponse analyzeResponse) {
+                if (null == analyzeResponse.getTokens() || analyzeResponse.getTokens().size() < 1) {
+                    String message = "elasticsearch analyze index fail.";
+                    LOGGER.error(message);
+                    throw new RuntimeException(message);
+                }
+            }
+
+            @Override
+            public void onFailure(final Exception e) {
+                LOGGER.error(e.getMessage());
+                throw new RuntimeException(e);
+            }
+        };
+        AnalyzeRequest analyzeRequest = new AnalyzeRequest(indexName).text("SkyWalking");
+        client.indices().analyzeAsync(analyzeRequest, RequestOptions.DEFAULT, listener);
+    }
+
    private void delete(RestHighLevelClient client, String indexName) throws IOException {
        DeleteIndexRequest request = new DeleteIndexRequest(indexName);
        AcknowledgedResponse deleteIndexResponse = client.indices().delete(request, RequestOptions.DEFAULT);
@ -207,4 +269,205 @@ public class RestHighLevelClientCase {
            throw new RuntimeException(message);
        }
    }
+
+    private void scroll(RestHighLevelClient client, String indexName) throws IOException {
+        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
+        sourceBuilder.query(QueryBuilders.matchAllQuery());
+
+        SearchRequest searchRequest = new SearchRequest();
+        searchRequest.indices(indexName);
+        searchRequest.source(sourceBuilder);
+        searchRequest.scroll(TimeValue.timeValueSeconds(30));
+        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
+
+        SearchScrollRequest scrollRequest = new SearchScrollRequest(searchResponse.getScrollId());
+        client.scroll(scrollRequest, RequestOptions.DEFAULT);
+    }
+
+    private void scrollAsync(RestHighLevelClient client, String indexName) throws IOException, InterruptedException {
+        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
+        sourceBuilder.query(QueryBuilders.matchAllQuery());
+
+        SearchRequest searchRequest = new SearchRequest();
+        searchRequest.indices(indexName);
+        searchRequest.source(sourceBuilder);
+        searchRequest.scroll(TimeValue.timeValueSeconds(30));
+        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
+
+        final CountDownLatch countDownLatch = new CountDownLatch(1);
+
+        ActionListener<SearchResponse> listener = new ActionListener<SearchResponse>() {
+            @Override
+            public void onResponse(final SearchResponse searchResponse) {
+                countDownLatch.countDown();
+            }
+
+            @Override
+            public void onFailure(final Exception e) {
+                countDownLatch.countDown();
+                LOGGER.error(e.getMessage());
+                throw new RuntimeException(e);
+            }
+        };
+
+        SearchScrollRequest scrollRequest = new SearchScrollRequest(searchResponse.getScrollId());
+        client.scrollAsync(scrollRequest, RequestOptions.DEFAULT, listener);
+
+        countDownLatch.await();
+    }
+
+    private void clearScroll(RestHighLevelClient client, String indexName) throws IOException {
+        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
+        sourceBuilder.query(QueryBuilders.matchAllQuery());
+
+        SearchRequest searchRequest = new SearchRequest();
+        searchRequest.indices(indexName);
+        searchRequest.source(sourceBuilder);
+        searchRequest.scroll(TimeValue.timeValueSeconds(30));
+        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
+
+        ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
+        clearScrollRequest.addScrollId(searchResponse.getScrollId());
+        client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
+    }
+
+    private void clearScrollAsync(RestHighLevelClient client,
+                                  String indexName) throws IOException {
+        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
+        sourceBuilder.query(QueryBuilders.matchAllQuery());
+
+        SearchRequest searchRequest = new SearchRequest();
+        searchRequest.indices(indexName);
+        searchRequest.source(sourceBuilder);
+        searchRequest.scroll(TimeValue.timeValueSeconds(30));
+        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
+
+        ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
+        clearScrollRequest.addScrollId(searchResponse.getScrollId());
+        client.clearScrollAsync(clearScrollRequest, RequestOptions.DEFAULT, new ActionListener<ClearScrollResponse>() {
+            @Override
+            public void onResponse(final ClearScrollResponse clearScrollResponse) {
+
+            }
+
+            @Override
+            public void onFailure(final Exception e) {
+                LOGGER.error(e.getMessage());
+                throw new RuntimeException(e);
+            }
+        });
+    }
+
+    private void searchTemplate(RestHighLevelClient client, String indexName) throws IOException {
+
+        SearchTemplateRequest searchTemplateRequest = new SearchTemplateRequest();
+        searchTemplateRequest.setRequest(new SearchRequest(indexName));
+        searchTemplateRequest.setScriptType(ScriptType.INLINE);
+
+        searchTemplateRequest.setScript("{\n" +
+                                            "  \"from\": \"{{from}}\",\n" +
+                                            "  \"size\": \"{{size}}\",\n" +
+                                            "  \"query\": {\n" +
+                                            "    \"term\": {\n" +
+                                            "      \"{{field}}\": {\n" +
+                                            "        \"value\": \"{{value}}\"\n" +
+                                            "      }\n" +
+                                            "    }\n" +
+                                            "  }\n" +
+                                            "}");
+
+        Map<String, Object> scriptParams = new HashMap<>();
+        scriptParams.put("field", "author");
+        scriptParams.put("value", "Marker");
+        scriptParams.put("size", 10);
+        scriptParams.put("from", 0);
+        searchTemplateRequest.setScriptParams(scriptParams);
+
+        SearchTemplateResponse searchTemplateResponse = client.searchTemplate(
+            searchTemplateRequest, RequestOptions.DEFAULT);
+        if (!(searchTemplateResponse.getResponse().getHits().totalHits > 0)) {
+            String message = "elasticsearch searchTemplateAsync data fail.";
+            LOGGER.error(message);
+            throw new RuntimeException(message);
+        }
+    }
+
+    private void searchTemplateAsync(RestHighLevelClient client, String indexName) throws IOException {
+
+        SearchTemplateRequest searchTemplateRequest = new SearchTemplateRequest();
+        searchTemplateRequest.setRequest(new SearchRequest(indexName));
+        searchTemplateRequest.setScriptType(ScriptType.INLINE);
+
+        searchTemplateRequest.setScript("{\n" +
+                                            "  \"from\": \"{{from}}\",\n" +
+                                            "  \"size\": \"{{size}}\",\n" +
+                                            "  \"query\": {\n" +
+                                            "    \"term\": {\n" +
+                                            "      \"{{field}}\": {\n" +
+                                            "        \"value\": \"{{value}}\"\n" +
+                                            "      }\n" +
+                                            "    }\n" +
+                                            "  }\n" +
+                                            "}");
+
+        Map<String, Object> scriptParams = new HashMap<>();
+        scriptParams.put("field", "author");
+        scriptParams.put("value", "Marker");
+        scriptParams.put("size", 10);
+        scriptParams.put("from", 0);
+        searchTemplateRequest.setScriptParams(scriptParams);
+
+        client.searchTemplateAsync(
+            searchTemplateRequest, RequestOptions.DEFAULT, new ActionListener<SearchTemplateResponse>() {
+                @Override
+                public void onResponse(final SearchTemplateResponse searchTemplateResponse) {
+                    if (!(searchTemplateResponse.getResponse().getHits().totalHits > 0)) {
+                        String message = "elasticsearch searchTemplateAsync data fail.";
+                        LOGGER.error(message);
+                        throw new RuntimeException(message);
+                    }
+                }
+
+                @Override
+                public void onFailure(final Exception e) {
+                    LOGGER.error(e);
+                    throw new RuntimeException();
+                }
+            });
+    }
+
+    private void deleteByQuery(RestHighLevelClient client, String indexName) throws IOException {
+        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
+        sourceBuilder.query(QueryBuilders.termQuery("author", "Marker1"));
+
+        DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(indexName);
+        deleteByQueryRequest.setQuery(sourceBuilder.query());
+        BulkByScrollResponse bulkByScrollResponse = client.deleteByQuery(
+            deleteByQueryRequest, RequestOptions.DEFAULT);
+        bulkByScrollResponse.getStatus();
+    }
+
+    private void deleteByQueryAsync(RestHighLevelClient client,
+                                    String indexName) throws IOException, InterruptedException {
+        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
+        sourceBuilder.query(QueryBuilders.termQuery("author", "Marker2"));
+
+        DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(indexName);
+        deleteByQueryRequest.setQuery(sourceBuilder.query());
+
+        client.deleteByQueryAsync(
+            deleteByQueryRequest, RequestOptions.DEFAULT, new ActionListener<BulkByScrollResponse>() {
+                @Override
+                public void onResponse(final BulkByScrollResponse bulkByScrollResponse) {
+
+                }
+
+                @Override
+                public void onFailure(final Exception e) {
+                    LOGGER.error(e);
+                    throw new RuntimeException();
+                }
+            }
+        );
+    }
 }