Fix delete- and update-by-query on indices without sequence numbers (#50077)

ywelsch · web-flow · commit 515707664ec7 · 2019-12-11T16:35:52.000+01:00
This fixes a bug where delete-by-query and update-by-query were not working on an older
index that did not have sequence numbers yet. The reason is that in a mixed-version 6.x/5.x
cluster, documents can have a primary term but miss a sequence number.
diff --git a/modules/reindex/src/main/java/org/elasticsearch/index/reindex/AsyncDeleteByQueryAction.java b/modules/reindex/src/main/java/org/elasticsearch/index/reindex/AsyncDeleteByQueryAction.java
@@ -25,6 +25,7 @@
 import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.client.ParentTaskAssigningClient;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.threadpool.ThreadPool;
 
@@ -33,18 +34,18 @@
  */
 public class AsyncDeleteByQueryAction extends AbstractAsyncBulkByScrollAction<DeleteByQueryRequest, TransportDeleteByQueryAction> {
 
-    private final boolean useSeqNoForCAS;
+    private final boolean allowSeqNoForCAS;
 
     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,
+            // always return version information, as sequence number information might not be available on an older index
+            true,
             // 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);
+        allowSeqNoForCAS = clusterState.nodes().getMinNodeVersion().onOrAfter(Version.V_6_7_0);
     }
 
     @Override
@@ -60,10 +61,11 @@ protected RequestWrapper<DeleteRequest> buildRequest(ScrollableHitSource.Hit doc
         delete.index(doc.getIndex());
         delete.type(doc.getType());
         delete.id(doc.getId());
-        if (useSeqNoForCAS) {
+        if (allowSeqNoForCAS && doc.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
             delete.setIfSeqNo(doc.getSeqNo());
             delete.setIfPrimaryTerm(doc.getPrimaryTerm());
         } else {
+            assert doc.getVersion() != -1 : "no version retrieved";
             delete.version(doc.getVersion());
         }
         return wrap(delete);
diff --git a/modules/reindex/src/main/java/org/elasticsearch/index/reindex/TransportUpdateByQueryAction.java b/modules/reindex/src/main/java/org/elasticsearch/index/reindex/TransportUpdateByQueryAction.java
@@ -38,6 +38,7 @@
 import org.elasticsearch.index.mapper.ParentFieldMapper;
 import org.elasticsearch.index.mapper.RoutingFieldMapper;
 import org.elasticsearch.index.mapper.TypeFieldMapper;
+import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.script.Script;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.tasks.Task;
@@ -88,18 +89,18 @@ protected void doExecute(UpdateByQueryRequest request, ActionListener<BulkByScro
      */
     static class AsyncIndexBySearchAction extends AbstractAsyncBulkByScrollAction<UpdateByQueryRequest, TransportUpdateByQueryAction> {
 
-        private final boolean useSeqNoForCAS;
+        private final boolean allowSeqNoForCAS;
 
         AsyncIndexBySearchAction(BulkByScrollTask task, Logger logger, ParentTaskAssigningClient client,
                 ThreadPool threadPool, TransportUpdateByQueryAction action, UpdateByQueryRequest request, 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,
+                // always return version information, as sequence number information might not be available on an older index
+                true,
                 // 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);
+            allowSeqNoForCAS = clusterState.nodes().getMinNodeVersion().onOrAfter(Version.V_6_7_0);
         }
 
         @Override
@@ -118,10 +119,11 @@ protected RequestWrapper<IndexRequest> buildRequest(ScrollableHitSource.Hit doc)
             index.type(doc.getType());
             index.id(doc.getId());
             index.source(doc.getSource(), doc.getXContentType());
-            if (useSeqNoForCAS) {
+            if (allowSeqNoForCAS && doc.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
                 index.setIfSeqNo(doc.getSeqNo());
                 index.setIfPrimaryTerm(doc.getPrimaryTerm());
             } else {
+                assert doc.getVersion() != -1 : "no version retrieved";
                 index.versionType(VersionType.INTERNAL);
                 index.version(doc.getVersion());
             }
diff --git a/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/RecoveryIT.java b/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/RecoveryIT.java
@@ -472,6 +472,62 @@ public void testUpdateDoc() throws Exception {
         }
     }
 
+    /** Ensure that we can always execute delete-by-query regardless of the version of cluster */
+    public void testDeleteByQuery() throws Exception {
+        final String index = "test_delete_by_query";
+        if (CLUSTER_TYPE == ClusterType.OLD) {
+            Settings.Builder settings = Settings.builder()
+                .put(IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), 1)
+                .put(IndexMetaData.INDEX_NUMBER_OF_REPLICAS_SETTING.getKey(), 2);
+            createIndex(index, settings.build());
+        }
+        for (int i = 0; i < 100; i++) {
+            Request indexDoc = new Request("POST", index + "/test");
+            indexDoc.setJsonEntity("{\"test\": \"test_" + randomInt(5) + "\"}");
+            client().performRequest(indexDoc);
+        }
+        client().performRequest(new Request("POST", index + "/_refresh"));
+        if (randomBoolean()) {
+            ensureGreen(index);
+        }
+        Request deleteByQuery = new Request("POST", index + "/_delete_by_query");
+        deleteByQuery.setJsonEntity("{\"query\": {\"term\": { \"test\": \"test_" + CLUSTER_TYPE.ordinal() + "\" }}}");
+        Map<String, Object> doc = entityAsMap(client().performRequest(deleteByQuery));
+        logger.info(doc);
+
+        if (randomBoolean()) {
+            syncedFlush(index);
+        }
+    }
+
+    /** Ensure that we can always execute delete-by-query regardless of the version of cluster */
+    public void testUpdateByQuery() throws Exception {
+        final String index = "test_update_by_query";
+        if (CLUSTER_TYPE == ClusterType.OLD) {
+            Settings.Builder settings = Settings.builder()
+                .put(IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), 1)
+                .put(IndexMetaData.INDEX_NUMBER_OF_REPLICAS_SETTING.getKey(), 2);
+            createIndex(index, settings.build());
+        }
+        for (int i = 0; i < 100; i++) {
+            Request indexDoc = new Request("POST", index + "/test");
+            indexDoc.setJsonEntity("{\"test\": \"test_" + randomInt(5) + "\"}");
+            client().performRequest(indexDoc);
+        }
+        client().performRequest(new Request("POST", index + "/_refresh"));
+        if (randomBoolean()) {
+            ensureGreen(index);
+        }
+        Request updateByQuery = new Request("POST", index + "/_update_by_query");
+        updateByQuery.setJsonEntity("{\"query\": {\"term\": { \"test\": \"test_" + CLUSTER_TYPE.ordinal() + "\" }}}");
+        Map<String, Object> doc = entityAsMap(client().performRequest(updateByQuery));
+        logger.info(doc);
+
+        if (randomBoolean()) {
+            syncedFlush(index);
+        }
+    }
+
     private void syncedFlush(String index) throws Exception {
         // We have to spin synced-flush requests here because we fire the global checkpoint sync for the last write operation.
         // A synced-flush request considers the global checkpoint sync as an going operation because it acquires a shard permit.
diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportUpdateFilterAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportUpdateFilterAction.java
@@ -31,6 +31,7 @@
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.engine.VersionConflictEngineException;
+import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.ml.MlMetaIndex;
@@ -109,7 +110,8 @@ private void indexUpdatedFilter(MlFilter filter, final long version, final long
                                     UpdateFilterAction.Request request,
                                     ActionListener<PutFilterAction.Response> listener) {
         IndexRequest indexRequest = new IndexRequest(MlMetaIndex.INDEX_NAME, MlMetaIndex.TYPE, filter.documentId());
-        if (clusterService.state().nodes().getMinNodeVersion().onOrAfter(Version.V_6_7_0)) {
+        if (clusterService.state().nodes().getMinNodeVersion().onOrAfter(Version.V_6_7_0) &&
+            seqNo != SequenceNumbers.UNASSIGNED_SEQ_NO) {
             indexRequest.setIfSeqNo(seqNo);
             indexRequest.setIfPrimaryTerm(primaryTerm);
         } else {
diff --git a/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authc/TokenService.java b/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authc/TokenService.java
@@ -884,7 +884,8 @@ private void innerRefresh(String tokenDocId, Authentication userAuth, ActionList
                                     client.prepareUpdate(SecurityIndexManager.SECURITY_INDEX_NAME, TYPE, tokenDocId)
                                         .setDoc("refresh_token", Collections.singletonMap("refreshed", true))
                                         .setRefreshPolicy(RefreshPolicy.WAIT_UNTIL);
-                                if (clusterService.state().nodes().getMinNodeVersion().onOrAfter(Version.V_6_7_0)) {
+                                if (clusterService.state().nodes().getMinNodeVersion().onOrAfter(Version.V_6_7_0) &&
+                                    response.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
                                     assert response.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO
                                         : "reading a token [" + tokenDocId + "] with no sequence number";
                                     assert response.getPrimaryTerm() != SequenceNumbers.UNASSIGNED_PRIMARY_TERM :
diff --git a/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/execution/ExecutionService.java b/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/execution/ExecutionService.java
@@ -39,6 +39,7 @@
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.index.engine.DocumentMissingException;
 import org.elasticsearch.index.engine.VersionConflictEngineException;
+import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.xpack.core.watcher.actions.ActionWrapper;
 import org.elasticsearch.xpack.core.watcher.actions.ActionWrapperResult;
 import org.elasticsearch.xpack.core.watcher.common.stats.Counters;
@@ -363,7 +364,8 @@ public void updateWatchStatus(Watch watch) throws IOException {
 
         UpdateRequest updateRequest = new UpdateRequest(Watch.INDEX, Watch.DOC_TYPE, watch.id());
         updateRequest.doc(source);
-        boolean useSeqNoForCAS = clusterService.state().nodes().getMinNodeVersion().onOrAfter(Version.V_6_7_0);
+        boolean useSeqNoForCAS = clusterService.state().nodes().getMinNodeVersion().onOrAfter(Version.V_6_7_0)
+            && watch.getSourceSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO;
         if (useSeqNoForCAS) {
             updateRequest.setIfSeqNo(watch.getSourceSeqNo());
             updateRequest.setIfPrimaryTerm(watch.getSourcePrimaryTerm());
diff --git a/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/ack/TransportAckWatchAction.java b/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/ack/TransportAckWatchAction.java
@@ -25,6 +25,7 @@
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -109,7 +110,8 @@ protected void masterOperation(AckWatchRequest request, ClusterState state,
 
                             UpdateRequest updateRequest = new UpdateRequest(Watch.INDEX, Watch.DOC_TYPE, request.getWatchId());
                             // this may reject this action, but prevents concurrent updates from a watch execution
-                            if (clusterService.state().nodes().getMinNodeVersion().onOrAfter(Version.V_6_7_0)) {
+                            if (clusterService.state().nodes().getMinNodeVersion().onOrAfter(Version.V_6_7_0) &&
+                                getResponse.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
                                 updateRequest.setIfSeqNo(getResponse.getSeqNo());
                                 updateRequest.setIfPrimaryTerm(getResponse.getPrimaryTerm());
                             } else {
