Fixed filters execution order and fix potential concurrency issue in filter chains

src/main/java/org/elasticsearch/action/support/TransportAction.java

@@ -27,6 +27,8 @@
 import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.threadpool.ThreadPool;
 
+import java.util.concurrent.atomic.AtomicInteger;
+
 import static org.elasticsearch.action.support.PlainActionFuture.newFuture;
 
 /**
@@ -146,12 +148,12 @@ public void run() {
 
     private class TransportActionFilterChain implements ActionFilterChain {
 
-        private volatile int index = 0;
+        private final AtomicInteger index = new AtomicInteger();
 
         @SuppressWarnings("unchecked")
         @Override
         public void continueProcessing(String action, ActionRequest actionRequest, ActionListener actionListener) {
-            int i = index++;
+            int i = index.getAndIncrement();
             try {
                 if (i < filters.length) {
                     filters[i].process(action, actionRequest, actionListener, this);

src/main/java/org/elasticsearch/rest/RestController.java

@@ -33,10 +33,9 @@
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Comparator;
+import java.util.concurrent.atomic.AtomicInteger;
 
-import static org.elasticsearch.rest.RestStatus.BAD_REQUEST;
-import static org.elasticsearch.rest.RestStatus.OK;
-import static org.elasticsearch.rest.RestStatus.FORBIDDEN;
+import static org.elasticsearch.rest.RestStatus.*;
 
 /**
  *
@@ -87,7 +86,7 @@ public synchronized void registerFilter(RestFilter preProcessor) {
         Arrays.sort(copy, new Comparator<RestFilter>() {
             @Override
             public int compare(RestFilter o1, RestFilter o2) {
-                return o2.order() - o1.order();
+                return Integer.compare(o1.order(), o2.order());
             }
         });
         filters = copy;
@@ -216,7 +215,7 @@ class ControllerFilterChain implements RestFilterChain {
 
         private final RestFilter executionFilter;
 
-        private volatile int index;
+        private final AtomicInteger index = new AtomicInteger();
 
         ControllerFilterChain(RestFilter executionFilter) {
             this.executionFilter = executionFilter;
@@ -225,8 +224,7 @@ class ControllerFilterChain implements RestFilterChain {
         @Override
         public void continueProcessing(RestRequest request, RestChannel channel) {
             try {
-                int loc = index;
-                index++;
+                int loc = index.getAndIncrement();
                 if (loc > filters.length) {
                     throw new ElasticsearchIllegalStateException("filter continueProcessing was called more than expected");
                 } else if (loc == filters.length) {

src/test/java/org/elasticsearch/action/support/TransportActionFilterChainTests.java

@@ -100,7 +100,7 @@ public int compare(ActionFilter o1, ActionFilter o2) {
         Collections.sort(testFiltersByLastExecution, new Comparator<TestFilter>() {
             @Override
             public int compare(TestFilter o1, TestFilter o2) {
-                return Long.compare(o1.lastExecution, o2.lastExecution);
+                return Integer.compare(o1.executionToken, o2.executionToken);
             }
         });
 
@@ -131,12 +131,7 @@ public void testTooManyContinueProcessing() throws ExecutionException, Interrupt
             @Override
             public void execute(final String action, final ActionRequest actionRequest, final ActionListener actionListener, final ActionFilterChain actionFilterChain) {
                 for (int i = 0; i <= additionalContinueCount; i++) {
-                    new Thread() {
-                        @Override
-                        public void run() {
-                            actionFilterChain.continueProcessing(action, actionRequest, actionListener);
-                        }
-                    }.start();
+                    actionFilterChain.continueProcessing(action, actionRequest, actionListener);
                 }
             }
         });
@@ -185,13 +180,15 @@ public void onFailure(Throwable e) {
         }
     }
 
-    private static class TestFilter implements ActionFilter {
+    private final AtomicInteger counter = new AtomicInteger();
+
+    private class TestFilter implements ActionFilter {
         private final int order;
         private final Callback callback;
 
         AtomicInteger runs = new AtomicInteger();
         volatile String lastActionName;
-        volatile long lastExecution = Long.MAX_VALUE; //the filters that don't run will go last in the sorted list
+        volatile int executionToken = Integer.MAX_VALUE; //the filters that don't run will go last in the sorted list
 
         TestFilter(int order, Callback callback) {
             this.order = order;
@@ -203,7 +200,7 @@ private static class TestFilter implements ActionFilter {
         public void process(String action, ActionRequest actionRequest, ActionListener actionListener, ActionFilterChain actionFilterChain) {
             this.runs.incrementAndGet();
             this.lastActionName = action;
-            this.lastExecution = System.nanoTime();
+            this.executionToken = counter.incrementAndGet();
             this.callback.execute(action, actionRequest, actionListener, actionFilterChain);
         }
 

src/test/java/org/elasticsearch/rest/FakeRestRequest.java

@@ -0,0 +1,98 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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.elasticsearch.rest;
+
+import org.elasticsearch.common.bytes.BytesReference;
+
+import java.util.HashMap;
+import java.util.Map;
+
+class FakeRestRequest extends RestRequest {
+
+    private final Map<String, String> headers;
+
+    FakeRestRequest() {
+        this(new HashMap<String, String>());
+    }
+
+    FakeRestRequest(Map<String, String> headers) {
+        this.headers = headers;
+    }
+
+    @Override
+    public Method method() {
+        return Method.GET;
+    }
+
+    @Override
+    public String uri() {
+        return "/";
+    }
+
+    @Override
+    public String rawPath() {
+        return "/";
+    }
+
+    @Override
+    public boolean hasContent() {
+        return false;
+    }
+
+    @Override
+    public boolean contentUnsafe() {
+        return false;
+    }
+
+    @Override
+    public BytesReference content() {
+        return null;
+    }
+
+    @Override
+    public String header(String name) {
+        return headers.get(name);
+    }
+
+    @Override
+    public Iterable<Map.Entry<String, String>> headers() {
+        return headers.entrySet();
+    }
+
+    @Override
+    public boolean hasParam(String key) {
+        return false;
+    }
+
+    @Override
+    public String param(String key) {
+        return null;
+    }
+
+    @Override
+    public String param(String key, String defaultValue) {
+        return null;
+    }
+
+    @Override
+    public Map<String, String> params() {
+        return null;
+    }
+}

src/test/java/org/elasticsearch/rest/HeadersCopyClientTests.java

@@ -35,7 +35,6 @@
 import org.elasticsearch.client.support.AbstractClient;
 import org.elasticsearch.client.support.AbstractClusterAdminClient;
 import org.elasticsearch.client.support.AbstractIndicesAdminClient;
-import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.test.ElasticsearchTestCase;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -332,75 +331,6 @@ private static void assertHeaders(ActionRequest<?> request, Map<String, String>
         }
     }
 
-    private static class FakeRestRequest extends RestRequest {
-
-        private final Map<String, String> headers;
-
-        private FakeRestRequest(Map<String, String> headers) {
-            this.headers = headers;
-        }
-
-        @Override
-        public Method method() {
-            return null;
-        }
-
-        @Override
-        public String uri() {
-            return null;
-        }
-
-        @Override
-        public String rawPath() {
-            return null;
-        }
-
-        @Override
-        public boolean hasContent() {
-            return false;
-        }
-
-        @Override
-        public boolean contentUnsafe() {
-            return false;
-        }
-
-        @Override
-        public BytesReference content() {
-            return null;
-        }
-
-        @Override
-        public String header(String name) {
-            return headers.get(name);
-        }
-
-        @Override
-        public Iterable<Map.Entry<String, String>> headers() {
-            return headers.entrySet();
-        }
-
-        @Override
-        public boolean hasParam(String key) {
-            return false;
-        }
-
-        @Override
-        public String param(String key) {
-            return null;
-        }
-
-        @Override
-        public String param(String key, String defaultValue) {
-            return null;
-        }
-
-        @Override
-        public Map<String, String> params() {
-            return null;
-        }
-    }
-
     private static class NoOpClient extends AbstractClient implements AdminClient {
 
         @Override