Return HTTP 409 (conflict) if get parallel put request (#1158)

(cherry picked from commit ea382b4)

src/main/java/com/amazon/opendistroforelasticsearch/security/dlic/rest/api/AbstractApiAction.java

@@ -39,6 +39,7 @@
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.index.engine.VersionConflictEngineException;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
 import org.elasticsearch.rest.RestChannel;
@@ -291,7 +292,11 @@ public OnSucessActionListener(RestChannel channel) {
 
 		@Override
 		public final void onFailure(Exception e) {
-			internalErrorResponse(channel, "Error "+e.getMessage());
+			if (ExceptionsHelper.unwrapCause(e) instanceof VersionConflictEngineException) {
+				conflict(channel, e.getMessage());
+			} else {
+				internalErrorResponse(channel, "Error "+e.getMessage());
+			}
 		}
 
 	}

src/test/java/com/amazon/opendistroforelasticsearch/security/dlic/rest/api/UserApiTest.java

@@ -57,6 +57,28 @@ public void testOpenDistroSecurityRoles() throws Exception {
         checkGeneralAccess(HttpStatus.SC_OK, "newuser", "newuser");
     }
 
+    @Test
+    public void testParallelPutRequests() throws Exception {
+
+        setup();
+
+        rh.keystore = "restapi/kirk-keystore.jks";
+        rh.sendAdminCertificate = true;
+
+        HttpResponse[] responses = rh.executeMultipleAsyncPutRequest(10, "/_opendistro/_security/api/internalusers/test1", "{\"password\":\"test1\"}");
+        boolean created = false;
+        for (HttpResponse response : responses) {
+            int sc = response.getStatusCode();
+            if (sc == HttpStatus.SC_CREATED) {
+                Assert.assertFalse(created);
+                created = true;
+            } else {
+                Assert.assertEquals(HttpStatus.SC_CONFLICT, sc);
+            }
+        }
+        deleteUser("test1");
+    }
+
     @Test
     public void testUserApi() throws Exception {
 

src/test/java/com/amazon/opendistroforelasticsearch/security/test/helper/rest/RestHelper.java

@@ -37,6 +37,9 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 
 import javax.net.ssl.SSLContext;
 
@@ -121,6 +124,18 @@ public String executeSimpleRequest(final String request) throws Exception {
 		}
 	}
 
+	public HttpResponse[] executeMultipleAsyncPutRequest(final int numOfRequests, final String request, String body) throws Exception {
+		final ExecutorService executorService = Executors.newFixedThreadPool(numOfRequests);
+		Future<HttpResponse>[] futures = new Future[numOfRequests];
+		for (int i = 0; i < numOfRequests; i++) {
+			futures[i] = executorService.submit(() -> executePutRequest(request, body, new Header[0]));
+		}
+		executorService.shutdown();
+		return Arrays.stream(futures)
+				.map(HttpResponse::from)
+				.toArray(s -> new HttpResponse[s]);
+	}
+
 	public HttpResponse executeGetRequest(final String request, Header... header) throws Exception {
 	    return executeRequest(new HttpGet(getHttpServerUri() + "/" + request), header);
 	}
@@ -259,7 +274,7 @@ protected final CloseableHttpClient getHTTPClient() throws Exception {
 	}
 
 
-	public class HttpResponse {
+	public static class HttpResponse {
 		private final CloseableHttpResponse inner;
 		private final String body;
 		private final Header[] header;
@@ -327,6 +342,13 @@ public String toString() {
                     + ", statusReason=" + statusReason + "]";
         }
 
+		private static HttpResponse from(Future<HttpResponse> future) {
+			try {
+				return future.get();
+			} catch (Exception e) {
+				throw new RuntimeException(e);
+			}
+		}
 	}