Fixed clearing scrolls after request

EinsamHauer · Jan 26, 2022 · c9bddef · c9bddef
1 parent 9fde21f
commit c9bddef
Show file tree

Hide file tree

Showing 2 changed files with 41 additions and 7 deletions.
diff --git a/pom.xml b/pom.xml
@@ -4,7 +4,7 @@
     <groupId>net.iponweb.disthene.reader</groupId>
     <artifactId>disthene-reader</artifactId>
     <packaging>jar</packaging>
-    <version>2.0.2</version>
+    <version>2.0.3</version>
     <name>disthene-reader</name>
     <url>http://maven.apache.org</url>
 

diff --git a/src/main/java/net/iponweb/disthene/reader/service/index/IndexService.java b/src/main/java/net/iponweb/disthene/reader/service/index/IndexService.java
@@ -7,10 +7,9 @@
 import org.apache.http.HttpHost;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.indices.settings.get.GetSettingsRequest;
-import org.elasticsearch.action.search.SearchRequest;
-import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.action.search.SearchScrollRequest;
+import org.elasticsearch.action.search.*;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.RestClient;
 import org.elasticsearch.client.RestHighLevelClient;
@@ -26,9 +25,7 @@
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
+import java.util.*;
 
 /**
  * @author Andrei Ivanov
@@ -79,8 +76,11 @@ private List<String> getPathsFromRegExs(String tenant, List<String> regExs, bool
                     .source(sourceBuilder)
                     .scroll(scroll);
 
+            final Set<String> scrollIds = new HashSet<>();
+
             SearchResponse response = client.search(request, RequestOptions.DEFAULT);
             String scrollId = response.getScrollId();
+            scrollIds.add(scrollId);
 
             // if total hits exceeds maximum - abort right away throwing an exception
             if (response.getHits().getTotalHits().value > indexConfiguration.getMaxPaths()) {
@@ -95,11 +95,17 @@ private List<String> getPathsFromRegExs(String tenant, List<String> regExs, bool
                     result.add(String.valueOf(hit.getSourceAsMap().get("path")));
                 }
 
+                logger.info(scrollId);
                 SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId).scroll(scroll);
                 response = client.scroll(scrollRequest, RequestOptions.DEFAULT);
                 scrollId = response.getScrollId();
+                scrollIds.add(scrollId);
+
                 hits = response.getHits();
+
             }
+
+            clearScrolls(scrollIds);
         }
 
         return result;
@@ -148,8 +154,11 @@ public String getPathsAsJsonArray(String tenant, String wildcard) throws TooMuch
                 .source(sourceBuilder)
                 .scroll(scroll);
 
+        final Set<String> scrollIds = new HashSet<>();
+
         SearchResponse response = client.search(request, RequestOptions.DEFAULT);
         String scrollId = response.getScrollId();
+        scrollIds.add(scrollId);
 
         // if total hits exceeds maximum - abort right away throwing an exception
         if (response.getHits().getTotalHits().value > indexConfiguration.getMaxPaths()) {
@@ -168,9 +177,13 @@ public String getPathsAsJsonArray(String tenant, String wildcard) throws TooMuch
             SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId).scroll(scroll);
             response = client.scroll(scrollRequest, RequestOptions.DEFAULT);
             scrollId = response.getScrollId();
+            scrollIds.add(scrollId);
+
             hits = response.getHits();
         }
 
+        clearScrolls(scrollIds);
+
         return "[" + joiner.join(paths) + "]";
     }
 
@@ -209,4 +222,25 @@ public void shutdown() {
             logger.error("Failed to close ES client: ", e);
         }
     }
+
+    private void clearScrolls(Iterable<String> scrollIds) {
+        ClearScrollRequest request = new ClearScrollRequest();
+        for (var scrollId : scrollIds) {
+            request.addScrollId(scrollId);
+        }
+
+        client.clearScrollAsync(request, RequestOptions.DEFAULT, new ActionListener<>() {
+            @Override
+            public void onResponse(ClearScrollResponse clearScrollResponse) {
+                // do nothing. We don't care
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                logger.warn("Failed to clear scroll with ids " + scrollIds, e);
+            }
+        });
+
+    }
+
 }