Rest Layer and Async Search Cleanup Management

src/main/java/com/amazon/opendistroforelasticsearch/search/async/context/active/AsyncSearchActiveStore.java

@@ -35,7 +35,7 @@ public class AsyncSearchActiveStore {
     private static Logger logger = LogManager.getLogger(AsyncSearchActiveStore.class);
     private volatile int maxRunningContext;
     public static final Setting<Integer> MAX_RUNNING_CONTEXT = Setting.intSetting(
-            "async_search.max_running_context", 100, 10, Setting.Property.Dynamic, Setting.Property.NodeScope);
+            "async_search.max_running_context", 100, 0, Setting.Property.Dynamic, Setting.Property.NodeScope);
 
     private final ConcurrentMapLong<AsyncSearchActiveContext> activeContexts = newConcurrentMapLongWithAggressiveConcurrency();
 

src/main/java/com/amazon/opendistroforelasticsearch/search/async/listener/AsyncSearchContextListener.java

@@ -71,4 +71,11 @@ default void onContextRunning(AsyncSearchContextId contextId) {
     default void onContextRejected(AsyncSearchContextId contextId) {
 
     }
+
+    /**
+     * @param contextId Executed when a running async search context is closed and has bypassed succeeded/failed state
+     */
+    default void onRunningContextClosed(AsyncSearchContextId contextId) {
+
+    }
 }

src/main/java/com/amazon/opendistroforelasticsearch/search/async/service/AsyncSearchService.java

@@ -418,7 +418,11 @@ private AsyncSearchStateMachine initStateMachine() {
                 (s, e) -> asyncSearchActiveStore.freeContext(e.asyncSearchContext().getContextId()),
                 (contextId, listener) -> listener.onContextPersistFailed(contextId), SearchResponsePersistFailedEvent.class));
 
-        for (AsyncSearchState state : EnumSet.of(PERSISTING, PERSISTED, PERSIST_FAILED, SUCCEEDED, FAILED, INIT, RUNNING)) {
+        stateMachine.registerTransition(new AsyncSearchTransition<>(RUNNING, CLOSED,
+                (s, e) -> asyncSearchActiveStore.freeContext(e.asyncSearchContext().getContextId()),
+                (contextId, listener) -> listener.onRunningContextClosed(contextId), SearchClosedEvent.class));
+
+        for (AsyncSearchState state : EnumSet.of(PERSISTING, PERSISTED, PERSIST_FAILED, SUCCEEDED, FAILED, INIT)) {
             stateMachine.registerTransition(new AsyncSearchTransition<>(state, CLOSED,
                     (s, e) -> asyncSearchActiveStore.freeContext(e.asyncSearchContext().getContextId()),
                     (contextId, listener) -> listener.onContextClosed(contextId), SearchClosedEvent.class));

src/main/java/com/amazon/opendistroforelasticsearch/search/async/stats/AsyncSearchCountStats.java

@@ -30,14 +30,16 @@ public class AsyncSearchCountStats implements Writeable, ToXContentFragment {
 
     private final long runningCount;
     private final long persistedCount;
+    private final long persistFailedCount;
     private final long completedCount;
     private final long failedCount;
     private final long throttledCount;
 
-    public AsyncSearchCountStats(long runningCount, long persistedCount,
-                                 long completedCount, long failedCount, long throttledCount) {
+    public AsyncSearchCountStats(long runningCount, long persistedCount, long completedCount, long failedCount, long throttledCount,
+                                 long persistFailedCount) {
         this.runningCount = runningCount;
         this.persistedCount = persistedCount;
+        this.persistFailedCount = persistFailedCount;
         this.completedCount = completedCount;
         this.failedCount = failedCount;
         this.throttledCount = throttledCount;
@@ -49,6 +51,7 @@ public AsyncSearchCountStats(StreamInput in) throws IOException {
         this.completedCount = in.readVLong();
         this.failedCount = in.readVLong();
         this.throttledCount = in.readVLong();
+        this.persistFailedCount = in.readVLong();
     }
 
     @Override
@@ -58,27 +61,30 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeVLong(this.completedCount);
         out.writeVLong(this.failedCount);
         out.writeVLong(this.throttledCount);
+        out.writeVLong(this.persistFailedCount);
     }
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        builder.startObject(Fields.ASYNC_SEARCH_STATUS);
+        builder.startObject(Fields.ASYNC_SEARCH_STATS);
         builder.field(Fields.RUNNING, runningCount);
         builder.field(Fields.PERSISTED, persistedCount);
         builder.field(Fields.FAILED, failedCount);
         builder.field(Fields.COMPLETED, completedCount);
         builder.field(Fields.REJECTED, throttledCount);
+        builder.field(Fields.PERSIST_FAILED, persistFailedCount);
         builder.endObject();
         return builder;
     }
 
     static final class Fields {
-        private static final String ASYNC_SEARCH_STATUS = "async_search_stats";
-        private static final String RUNNING = "async_search_running_current";
-        private static final String PERSISTED = "async_search_persisted";
-        private static final String FAILED = "async_search_failed";
-        private static final String COMPLETED = "async_search_completed";
-        private static final String REJECTED = "async_search_rejected";
+        private static final String ASYNC_SEARCH_STATS = "asynchronous_search_stats";
+        private static final String RUNNING = "running_current";
+        private static final String PERSISTED = "persisted";
+        private static final String PERSIST_FAILED = "persist_failed";
+        private static final String FAILED = "failed";
+        private static final String COMPLETED = "completed";
+        private static final String REJECTED = "rejected";
     }
 
     public long getRunningCount() {

src/main/java/com/amazon/opendistroforelasticsearch/search/async/stats/InternalAsyncSearchStats.java

@@ -35,6 +35,11 @@ public void onContextPersisted(AsyncSearchContextId asyncSearchContextId) {
         countStatsHolder.persistedAsyncSearchCount.inc();
     }
 
+    @Override
+    public void onContextPersistFailed(AsyncSearchContextId contextId) {
+        countStatsHolder.persistFailedAsyncSearchCount.inc();
+    }
+
     @Override
     public void onContextRunning(AsyncSearchContextId context) {
         countStatsHolder.runningAsyncSearchCount.inc();
@@ -45,6 +50,11 @@ public void onContextRejected(AsyncSearchContextId contextId) {
         countStatsHolder.rejectedAsyncSearchCount.inc();
     }
 
+    @Override
+    public void onRunningContextClosed(AsyncSearchContextId contextId) {
+        countStatsHolder.runningAsyncSearchCount.dec();
+    }
+
     @Override
     public void onContextCompleted(AsyncSearchContextId context) {
         countStatsHolder.completedAsyncSearchCount.inc();
@@ -58,13 +68,15 @@ public AsyncSearchStats stats(DiscoveryNode node) {
     static final class CountStatsHolder {
         final CounterMetric runningAsyncSearchCount = new CounterMetric();
         final CounterMetric persistedAsyncSearchCount = new CounterMetric();
+        final CounterMetric persistFailedAsyncSearchCount = new CounterMetric();
         final CounterMetric failedAsyncSearchCount = new CounterMetric();
         final CounterMetric completedAsyncSearchCount = new CounterMetric();
         final CounterMetric rejectedAsyncSearchCount = new CounterMetric();
 
         public AsyncSearchCountStats countStats() {
             return new AsyncSearchCountStats(runningAsyncSearchCount.count(), persistedAsyncSearchCount.count(),
-                    completedAsyncSearchCount.count(), failedAsyncSearchCount.count(), rejectedAsyncSearchCount.count());
+                    completedAsyncSearchCount.count(), failedAsyncSearchCount.count(), rejectedAsyncSearchCount.count(),
+                    persistFailedAsyncSearchCount.count());
         }
     }
 }