Expose Sequence Number based Optimistic Concurrency Control in the rest layer (#36721)

Relates #36148
Relates #10708

Author: bleskes

rest-api-spec/src/main/resources/rest-api-spec/api/delete.json

@@ -44,6 +44,14 @@
           "type" : "time",
           "description" : "Explicit operation timeout"
         },
+        "if_seq_no_match" : {
+          "type" : "number",
+          "description" : "only perform the delete operation if the last operation that has changed the document has the specified sequence number"
+        },
+        "if_primary_term_match" : {
+          "type" : "number",
+          "description" : "only perform the delete operation if the last operation that has changed the document has the specified primary term"
+        },
         "version" : {
           "type" : "number",
           "description" : "Explicit version number for concurrency control"

rest-api-spec/src/main/resources/rest-api-spec/api/index.json

@@ -58,6 +58,14 @@
           "options" : ["internal", "external", "external_gte", "force"],
           "description" : "Specific version type"
         },
+        "if_seq_no_match" : {
+          "type" : "number",
+          "description" : "only perform the index operation if the last operation that has changed the document has the specified sequence number"
+        },
+        "if_primary_term_match" : {
+          "type" : "number",
+          "description" : "only perform the index operation if the last operation that has changed the document has the specified primary term"
+        },
         "pipeline" : {
           "type" : "string",
           "description" : "The pipeline id to preprocess incoming documents with"

rest-api-spec/src/main/resources/rest-api-spec/test/index/30_cas.yml

@@ -0,0 +1,50 @@
+---
+"Compare And Swap Sequence Numbers":
+
+ - skip:
+      version: " - 6.5.99"
+      reason:  cas ops are introduced in 6.6.0
+
+ - do:
+      index:
+          index:  test_1
+          id:     1
+          body:   { foo: bar }
+ - match:   { _version: 1}
+ - set: { _seq_no: seqno }
+ - set: { _primary_term: primary_term }
+
+ - do:
+      get:
+          index:  test_1
+          id:     1
+ - match: { _seq_no: $seqno }
+ - match: { _primary_term: $primary_term }
+
+ - do:
+      catch:        conflict
+      index:
+          index:            test_1
+          id:               1
+          if_seq_no_match:       10000
+          if_primary_term_match: $primary_term
+          body:   { foo: bar2 }
+
+ - do:
+      catch:       conflict
+      index:
+          index:            test_1
+          id:               1
+          if_seq_no_match:       $seqno
+          if_primary_term_match: 1000
+          body:   { foo: bar2 }
+
+ - do:
+      index:
+          index:            test_1
+          id:               1
+          if_seq_no_match:       $seqno
+          if_primary_term_match: $primary_term
+          body:   { foo: bar2 }
+
+ - match: { _version: 2 }

server/src/main/java/org/elasticsearch/action/get/GetResponse.java

@@ -91,7 +91,7 @@ public long getVersion() {
     }
 
     /**
-     * The sequence number assigned to the last operation to have changed this document, if found.
+     * The sequence number assigned to the last operation that has changed this document, if found.
      */
     public long getSeqNo() {
         return getResult.getSeqNo();

server/src/main/java/org/elasticsearch/index/get/GetResult.java

@@ -131,7 +131,7 @@ public long getVersion() {
     }
 
     /**
-     * The sequence number assigned to the last operation to have changed this document, if found.
+     * The sequence number assigned to the last operation that has changed this document, if found.
      */
     public long getSeqNo() {
         return seqNo;

server/src/main/java/org/elasticsearch/rest/action/document/RestDeleteAction.java

@@ -54,6 +54,10 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         deleteRequest.setRefreshPolicy(request.param("refresh"));
         deleteRequest.version(RestActions.parseVersion(request));
         deleteRequest.versionType(VersionType.fromString(request.param("version_type"), deleteRequest.versionType()));
+        deleteRequest.setIfMatch(
+            request.paramAsLong("if_seq_no_match", deleteRequest.ifSeqNoMatch()),
+            request.paramAsLong("if_primary_term_match", deleteRequest.ifPrimaryTermMatch())
+        );
 
         String waitForActiveShards = request.param("wait_for_active_shards");
         if (waitForActiveShards != null) {

server/src/main/java/org/elasticsearch/rest/action/document/RestIndexAction.java

@@ -79,6 +79,10 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         indexRequest.setRefreshPolicy(request.param("refresh"));
         indexRequest.version(RestActions.parseVersion(request));
         indexRequest.versionType(VersionType.fromString(request.param("version_type"), indexRequest.versionType()));
+        indexRequest.ifMatch(
+            request.paramAsLong("if_seq_no_match", indexRequest.ifSeqNoMatch()),
+            request.paramAsLong("if_primary_term_match", indexRequest.ifPrimaryTermMatch())
+        );
         String sOpType = request.param("op_type");
         String waitForActiveShards = request.param("wait_for_active_shards");
         if (waitForActiveShards != null) {