Remove support for internal versioning for concurrency control

client/rest-high-level/src/main/java/org/elasticsearch/client/WatcherRequestConverters.java

@@ -70,7 +70,6 @@ static Request putWatch(PutWatchRequest putWatchRequest) {
 
         Request request = new Request(HttpPut.METHOD_NAME, endpoint);
         RequestConverters.Params params = new RequestConverters.Params(request)
-            .withVersion(putWatchRequest.getVersion())
             .withIfSeqNo(putWatchRequest.ifSeqNo())
             .withIfPrimaryTerm(putWatchRequest.ifPrimaryTerm());
         if (putWatchRequest.isActive() == false) {

client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/PutWatchRequest.java

@@ -21,7 +21,6 @@
 import org.elasticsearch.client.Validatable;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
-import org.elasticsearch.common.lucene.uid.Versions;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.seqno.SequenceNumbers;
 
@@ -43,11 +42,9 @@ public final class PutWatchRequest implements Validatable {
     private final BytesReference source;
     private final XContentType xContentType;
     private boolean active = true;
-    private long version = Versions.MATCH_ANY;
     private long ifSeqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
     private long ifPrimaryTerm = UNASSIGNED_PRIMARY_TERM;
 
-
     public PutWatchRequest(String id, BytesReference source, XContentType xContentType) {
         Objects.requireNonNull(id, "watch id is missing");
         if (isValidId(id) == false) {
@@ -95,14 +92,6 @@ public XContentType xContentType() {
         return xContentType;
     }
 
-    public long getVersion() {
-        return version;
-    }
-
-    public void setVersion(long version) {
-        this.version = version;
-    }
-
     /**
      * only performs this put request if the watch's last modification was assigned the given
      * sequence number. Must be used in combination with {@link #setIfPrimaryTerm(long)}

client/rest-high-level/src/test/java/org/elasticsearch/client/RequestConvertersTests.java

@@ -768,8 +768,6 @@ public void testUpdate() throws IOException {
             }
         }
         setRandomWaitForActiveShards(updateRequest::waitForActiveShards, expectedParams);
-        setRandomVersion(updateRequest, expectedParams);
-        setRandomVersionType(updateRequest::versionType, expectedParams);
         setRandomIfSeqNoAndTerm(updateRequest, new HashMap<>()); // if* params are passed in the body
         if (randomBoolean()) {
             int retryOnConflict = randomIntBetween(0, 5);
@@ -911,14 +909,7 @@ public void testBulk() throws IOException {
             if (randomBoolean()) {
                 docWriteRequest.routing(randomAlphaOfLength(10));
             }
-            if (randomBoolean()) {
-                if (randomBoolean()) {
-                    docWriteRequest.version(randomNonNegativeLong());
-                }
-                if (randomBoolean()) {
-                    docWriteRequest.versionType(randomFrom(VersionType.values()));
-                }
-            } else if (randomBoolean()) {
+            if (opType != DocWriteRequest.OpType.UPDATE && randomBoolean()) {
                 docWriteRequest.setIfSeqNo(randomNonNegativeLong());
                 docWriteRequest.setIfPrimaryTerm(randomLongBetween(1, 200));
             }

client/rest-high-level/src/test/java/org/elasticsearch/client/WatcherRequestConvertersTests.java

@@ -29,8 +29,8 @@
 import org.elasticsearch.client.watcher.DeactivateWatchRequest;
 import org.elasticsearch.client.watcher.DeleteWatchRequest;
 import org.elasticsearch.client.watcher.ExecuteWatchRequest;
-import org.elasticsearch.client.watcher.PutWatchRequest;
 import org.elasticsearch.client.watcher.GetWatchRequest;
+import org.elasticsearch.client.watcher.PutWatchRequest;
 import org.elasticsearch.client.watcher.StartWatchServiceRequest;
 import org.elasticsearch.client.watcher.StopWatchServiceRequest;
 import org.elasticsearch.client.watcher.WatcherStatsRequest;
@@ -88,9 +88,12 @@ public void testPutWatch() throws Exception {
         }
 
         if (randomBoolean()) {
-            long version = randomLongBetween(10, 100);
-            putWatchRequest.setVersion(version);
-            expectedParams.put("version", String.valueOf(version));
+            long seqNo = randomNonNegativeLong();
+            long ifPrimaryTerm = randomLongBetween(1, 200);
+            putWatchRequest.setIfSeqNo(seqNo);
+            putWatchRequest.setIfPrimaryTerm(ifPrimaryTerm);
+            expectedParams.put("if_seq_no", String.valueOf(seqNo));
+            expectedParams.put("if_primary_term", String.valueOf(ifPrimaryTerm));
         }
 
         Request request = WatcherRequestConverters.putWatch(putWatchRequest);

client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/CRUDDocumentationIT.java

@@ -170,7 +170,6 @@ public void testIndex() throws Exception {
             // tag::index-response
             String index = indexResponse.getIndex();
             String id = indexResponse.getId();
-            long version = indexResponse.getVersion();
             if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
                 // <1>
             } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
@@ -220,7 +219,8 @@ public void testIndex() throws Exception {
             IndexRequest request = new IndexRequest("posts")
                 .id("1")
                 .source("field", "value")
-                .version(1);
+                .setIfSeqNo(10L)
+                .setIfPrimaryTerm(20);
             try {
                 IndexResponse response = client.index(request, RequestOptions.DEFAULT);
             } catch(ElasticsearchException e) {
@@ -432,7 +432,8 @@ public void testUpdate() throws Exception {
             // tag::update-conflict
             UpdateRequest request = new UpdateRequest("posts", "1")
                     .doc("field", "value")
-                    .version(1);
+                    .setIfSeqNo(101L)
+                    .setIfPrimaryTerm(200L);
             try {
                 UpdateResponse updateResponse = client.update(
                         request, RequestOptions.DEFAULT);
@@ -499,9 +500,10 @@ public void testUpdate() throws Exception {
             request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL); // <1>
             request.setRefreshPolicy("wait_for");                            // <2>
             // end::update-request-refresh
-            // tag::update-request-version
-            request.version(2); // <1>
-            // end::update-request-version
+            // tag::update-request-cas
+            request.setIfSeqNo(2L); // <1>
+            request.setIfPrimaryTerm(1L); // <2>
+            // end::update-request-request-cas
             // tag::update-request-detect-noop
             request.detectNoop(false); // <1>
             // end::update-request-detect-noop
@@ -630,7 +632,7 @@ public void testDelete() throws Exception {
             // tag::delete-conflict
             try {
                 DeleteResponse deleteResponse = client.delete(
-                        new DeleteRequest("posts", "1").version(2),
+                    new DeleteRequest("posts", "1").setIfSeqNo(100).setIfPrimaryTerm(2),
                         RequestOptions.DEFAULT);
             } catch (ElasticsearchException exception) {
                 if (exception.status() == RestStatus.CONFLICT) {

docs/java-rest/high-level/document/update.asciidoc

@@ -140,9 +140,10 @@ include-tagged::{doc-tests-file}[{api}-request-source-exclude]
 
 ["source","java",subs="attributes,callouts,macros"]
 --------------------------------------------------
-include-tagged::{doc-tests-file}[{api}-request-version]
+include-tagged::{doc-tests-file}[{api}-request-cas]
 --------------------------------------------------
-<1> Version
+<1> ifSeqNo
+<2> ifPrimaryTerm
 
 ["source","java",subs="attributes,callouts,macros"]
 --------------------------------------------------

docs/reference/migration/migrate_7_0/api.asciidoc

@@ -2,6 +2,22 @@
 [[breaking_70_api_changes]]
 === API changes
 
+[float]
+==== Internal Versioning is no longer supported for optimistic concurrency control
+
+Elasticsearch maintains a numeric version field for each document it stores. That field
+is incremented by one with every change to the document. Until 7.0.0 the API allowed using
+that field for optimistic concurrency control, i.e., making a write operation conditional
+on the current document version. Sadly, that approach is flawed because the value of the
+version doesn't always uniquely represent a change to the document. If a primary fails
+while handling a write operation, it may expose a version that will then be reused by the
+new primary.
+
+Due to that issue, internal versioning can no longer be used and is replaced by a new
+method based on sequence numbers. See <<optimistic-concurrency-control>> for more details.
+
+Note that the `external` versioning type is still fully supported.
+
 [float]
 ==== Camel case and underscore parameters deprecated in 6.x have been removed
 A number of duplicate parameters deprecated in 6.x have been removed from

modules/reindex/src/main/java/org/elasticsearch/index/reindex/AsyncDeleteByQueryAction.java

@@ -20,11 +20,9 @@
 package org.elasticsearch.index.reindex;
 
 import org.apache.logging.log4j.Logger;
-import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.client.ParentTaskAssigningClient;
-import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.threadpool.ThreadPool;
 
@@ -33,18 +31,10 @@
  */
 public class AsyncDeleteByQueryAction extends AbstractAsyncBulkByScrollAction<DeleteByQueryRequest, TransportDeleteByQueryAction> {
 
-    private final boolean useSeqNoForCAS;
-
     public AsyncDeleteByQueryAction(BulkByScrollTask task, Logger logger, ParentTaskAssigningClient client,
                                     ThreadPool threadPool, TransportDeleteByQueryAction action, DeleteByQueryRequest request,
-                                    ScriptService scriptService, ClusterState clusterState, ActionListener<BulkByScrollResponse> listener) {
-        super(task,
-            // not all nodes support sequence number powered optimistic concurrency control, we fall back to version
-            clusterState.nodes().getMinNodeVersion().onOrAfter(Version.V_6_7_0) == false,
-            // all nodes support sequence number powered optimistic concurrency control and we can use it
-            clusterState.nodes().getMinNodeVersion().onOrAfter(Version.V_6_7_0),
-            logger, client, threadPool, action, request, listener);
-        useSeqNoForCAS = clusterState.nodes().getMinNodeVersion().onOrAfter(Version.V_6_7_0);
+                                    ScriptService scriptService, ActionListener<BulkByScrollResponse> listener) {
+        super(task, false, true, logger, client, threadPool, action, request, listener);
     }
 
     @Override
@@ -60,12 +50,8 @@ protected RequestWrapper<DeleteRequest> buildRequest(ScrollableHitSource.Hit doc
         delete.index(doc.getIndex());
         delete.type(doc.getType());
         delete.id(doc.getId());
-        if (useSeqNoForCAS) {
-            delete.setIfSeqNo(doc.getSeqNo());
-            delete.setIfPrimaryTerm(doc.getPrimaryTerm());
-        } else {
-            delete.version(doc.getVersion());
-        }
+        delete.setIfSeqNo(doc.getSeqNo());
+        delete.setIfPrimaryTerm(doc.getPrimaryTerm());
         return wrap(delete);
     }
 

modules/reindex/src/main/java/org/elasticsearch/index/reindex/TransportDeleteByQueryAction.java

@@ -61,7 +61,7 @@ public void doExecute(Task task, DeleteByQueryRequest request, ActionListener<Bu
                 ClusterState state = clusterService.state();
                 ParentTaskAssigningClient assigningClient = new ParentTaskAssigningClient(client, clusterService.localNode(),
                     bulkByScrollTask);
-                new AsyncDeleteByQueryAction(bulkByScrollTask, logger, assigningClient, threadPool, this, request, scriptService, state,
+                new AsyncDeleteByQueryAction(bulkByScrollTask, logger, assigningClient, threadPool, this, request, scriptService,
                     listener).start();
             }
         );

modules/reindex/src/main/java/org/elasticsearch/index/reindex/TransportUpdateByQueryAction.java

@@ -31,7 +31,6 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.index.VersionType;
 import org.elasticsearch.index.mapper.IdFieldMapper;
 import org.elasticsearch.index.mapper.IndexFieldMapper;
 import org.elasticsearch.index.mapper.RoutingFieldMapper;
@@ -113,13 +112,8 @@ protected RequestWrapper<IndexRequest> buildRequest(ScrollableHitSource.Hit doc)
             index.type(doc.getType());
             index.id(doc.getId());
             index.source(doc.getSource(), doc.getXContentType());
-            if (useSeqNoForCAS) {
-                index.setIfSeqNo(doc.getSeqNo());
-                index.setIfPrimaryTerm(doc.getPrimaryTerm());
-            } else {
-                index.versionType(VersionType.INTERNAL);
-                index.version(doc.getVersion());
-            }
+            index.setIfSeqNo(doc.getSeqNo());
+            index.setIfPrimaryTerm(doc.getPrimaryTerm());
             index.setPipeline(mainRequest.getPipeline());
             return wrap(index);
         }

modules/reindex/src/test/java/org/elasticsearch/index/reindex/UpdateByQueryWhileModifyingTests.java

@@ -67,15 +67,15 @@ public void testUpdateWhileReindexing() throws Exception {
                 IndexRequestBuilder index = client().prepareIndex("test", "test", "test").setSource("test", value.get())
                         .setRefreshPolicy(IMMEDIATE);
                 /*
-                 * Update by query increments the version number so concurrent
+                 * Update by query changes the document so concurrent
                  * indexes might get version conflict exceptions so we just
                  * blindly retry.
                  */
                 int attempts = 0;
                 while (true) {
                     attempts++;
                     try {
-                        index.setVersion(get.getVersion()).get();
+                        index.setIfSeqNo(get.getSeqNo()).setIfPrimaryTerm(get.getPrimaryTerm()).get();
                         break;
                     } catch (VersionConflictEngineException e) {
                         if (attempts >= MAX_ATTEMPTS) {

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

@@ -70,15 +70,6 @@
         "if_primary_term" : {
           "type" : "number",
           "description" : "only perform the update 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"
-        },
-        "version_type": {
-          "type": "enum",
-          "options": ["internal", "force"],
-          "description": "Specific version type"
         }
       }
     },

rest-api-spec/src/main/resources/rest-api-spec/test/delete/20_internal_version.yml → rest-api-spec/src/main/resources/rest-api-spec/test/delete/20_cas.yml

@@ -11,19 +11,21 @@
           id:      1
           body:    { foo: bar }
 
- - match:   { _version: 1}
+ - match:   { _seq_no: 0 }
 
  - do:
       catch:       conflict
       delete:
           index:   test_1
           id:      1
-          version: 2
+          if_seq_no: 2
+          if_primary_term: 1
 
  - do:
       delete:
           index:   test_1
           id:      1
-          version: 1
+          if_seq_no: 0
+          if_primary_term: 1
 
- - match: { _version: 2 }
+ - match: { _seq_no: 1 }