Clear ES index scrolls if the result set is less than the requested limit.
Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/f89bbdbc Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/f89bbdbc Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/f89bbdbc Branch: refs/heads/hotfix-2.0.0 Commit: f89bbdbc0e8b802be38d938a3c4539d6cc38f01e Parents: e3b467d Author: Michael Russo <[email protected]> Authored: Tue Mar 29 13:10:11 2016 -0700 Committer: Michael Russo <[email protected]> Committed: Tue Mar 29 13:10:11 2016 -0700 ---------------------------------------------------------------------- .../index/impl/EsEntityIndexImpl.java | 32 +++++++++++++++----- 1 file changed, 25 insertions(+), 7 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/usergrid/blob/f89bbdbc/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java ---------------------------------------------------------------------- diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java index 9add426..57d263e 100644 --- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java +++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexImpl.java @@ -54,9 +54,7 @@ import org.elasticsearch.action.admin.indices.alias.IndicesAliasesResponse; import org.elasticsearch.action.admin.indices.create.CreateIndexResponse; import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse; import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse; -import org.elasticsearch.action.search.SearchRequestBuilder; -import org.elasticsearch.action.search.SearchResponse; -import org.elasticsearch.action.search.SearchScrollRequestBuilder; +import org.elasticsearch.action.search.*; import org.elasticsearch.client.AdminClient; import org.elasticsearch.common.settings.ImmutableSettings; import org.elasticsearch.common.settings.Settings; @@ -390,6 +388,7 @@ public class EsEntityIndexImpl implements AliasedEntityIndex { SearchResponse searchResponse; + String esScrollCursor = null; if ( query.getCursor() == null ) { SearchRequestBuilder srb = esProvider.getClient().prepareSearch( alias.getReadAlias() ) @@ -474,7 +473,7 @@ public class EsEntityIndexImpl implements AliasedEntityIndex { } //now get the cursor from the map and validate - final String esScrollCursor = mapManager.getString( userCursorString ); + esScrollCursor = mapManager.getString( userCursorString ); Preconditions.checkArgument(esScrollCursor != null, "Could not find a cursor for the value '{}' ", esScrollCursor); @@ -502,11 +501,11 @@ public class EsEntityIndexImpl implements AliasedEntityIndex { failureMonitor.success(); } - return parseResults(searchResponse, query); + return parseResults(searchResponse, query, esScrollCursor ); } - private CandidateResults parseResults( final SearchResponse searchResponse, final Query query ) { + private CandidateResults parseResults(final SearchResponse searchResponse, final Query query, String previousCursor) { final SearchHits searchHits = searchResponse.getHits(); final SearchHit[] hits = searchHits.getHits(); @@ -546,6 +545,25 @@ public class EsEntityIndexImpl implements AliasedEntityIndex { candidateResults.setCursor( userCursorString ); logger.debug(" User cursor = {}, Cursor = {} ", userCursorString, esScrollCursor); + }else{ + + // just execute this and not care about the response + logger.info("Candidate results size {} is less than limit {}, clearing next scroll ID {}", + candidateResults.size(), query.getLimit(), searchResponse.getScrollId()); + + ClearScrollRequestBuilder builder = esProvider.getClient().prepareClearScroll() + .addScrollId(searchResponse.getScrollId()); + + if( previousCursor != null){ + + logger.info("Candidate results size {} is less than limit {}, clearing previous scroll ID {}", + candidateResults.size(), query.getLimit(), previousCursor); + + builder.addScrollId(previousCursor); + } + + builder.execute().actionGet(); + } return candidateResults; @@ -632,7 +650,7 @@ public class EsEntityIndexImpl implements AliasedEntityIndex { failureMonitor.success(); - return parseResults(searchResponse, new Query()); + return parseResults(searchResponse, new Query(), null); }
