Merge branch 'ccr' into ccr_expand_stats_api_response

* ccr:
  Fix ShardFollowNodeTask.Status equals and hash code (elastic#33189)
  Make soft-deletes settings final (elastic#33172)
  Only fetch mapping updates when necessary (elastic#33182)

Author: jasontedor

server/src/main/java/org/elasticsearch/index/IndexSettings.java

@@ -242,7 +242,7 @@ public final class IndexSettings {
      * Specifies if the index should use soft-delete instead of hard-delete for update/delete operations.
      */
     public static final Setting<Boolean> INDEX_SOFT_DELETES_SETTING =
-        Setting.boolSetting("index.soft_deletes.enabled", true, Property.IndexScope);
+        Setting.boolSetting("index.soft_deletes.enabled", true, Property.IndexScope, Property.Final);
 
     /**
      * Controls how many soft-deleted documents will be kept around before being merged away. Keeping more deleted

server/src/test/java/org/elasticsearch/index/IndexSettingsTests.java

@@ -553,4 +553,12 @@ public void testQueryDefaultField() {
         );
         assertThat(index.getDefaultFields(), equalTo(Arrays.asList("body", "title")));
     }
+
+    public void testUpdateSoftDeletesFails() {
+        IndexScopedSettings settings = new IndexScopedSettings(Settings.EMPTY, IndexScopedSettings.BUILT_IN_INDEX_SETTINGS);
+        IllegalArgumentException error = expectThrows(IllegalArgumentException.class, () ->
+            settings.updateSettings(Settings.builder().put("index.soft_deletes.enabled", randomBoolean()).build(),
+                Settings.builder(), Settings.builder(), "index"));
+        assertThat(error.getMessage(), equalTo("final index setting [index.soft_deletes.enabled], not updateable"));
+    }
 }

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardChangesAction.java

@@ -161,10 +161,10 @@ public String toString() {
 
     public static final class Response extends ActionResponse {
 
-        private long indexMetadataVersion;
+        private long mappingVersion;
 
-        public long getIndexMetadataVersion() {
-            return indexMetadataVersion;
+        public long getMappingVersion() {
+            return mappingVersion;
         }
 
         private long globalCheckpoint;
@@ -188,8 +188,8 @@ public Translog.Operation[] getOperations() {
         Response() {
         }
 
-        Response(final long indexMetadataVersion, final long globalCheckpoint, final long maxSeqNo, final Translog.Operation[] operations) {
-            this.indexMetadataVersion = indexMetadataVersion;
+        Response(final long mappingVersion, final long globalCheckpoint, final long maxSeqNo, final Translog.Operation[] operations) {
+            this.mappingVersion = mappingVersion;
             this.globalCheckpoint = globalCheckpoint;
             this.maxSeqNo = maxSeqNo;
             this.operations = operations;
@@ -198,7 +198,7 @@ public Translog.Operation[] getOperations() {
         @Override
         public void readFrom(final StreamInput in) throws IOException {
             super.readFrom(in);
-            indexMetadataVersion = in.readVLong();
+            mappingVersion = in.readVLong();
             globalCheckpoint = in.readZLong();
             maxSeqNo = in.readZLong();
             operations = in.readArray(Translog.Operation::readOperation, Translog.Operation[]::new);
@@ -207,7 +207,7 @@ public void readFrom(final StreamInput in) throws IOException {
         @Override
         public void writeTo(final StreamOutput out) throws IOException {
             super.writeTo(out);
-            out.writeVLong(indexMetadataVersion);
+            out.writeVLong(mappingVersion);
             out.writeZLong(globalCheckpoint);
             out.writeZLong(maxSeqNo);
             out.writeArray(Translog.Operation::writeOperation, operations);
@@ -218,15 +218,15 @@ public boolean equals(final Object o) {
             if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
             final Response that = (Response) o;
-            return indexMetadataVersion == that.indexMetadataVersion &&
+            return mappingVersion == that.mappingVersion &&
                     globalCheckpoint == that.globalCheckpoint &&
                     maxSeqNo == that.maxSeqNo &&
                     Arrays.equals(operations, that.operations);
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(indexMetadataVersion, globalCheckpoint, maxSeqNo, Arrays.hashCode(operations));
+            return Objects.hash(mappingVersion, globalCheckpoint, maxSeqNo, Arrays.hashCode(operations));
         }
     }
 
@@ -252,15 +252,15 @@ protected Response shardOperation(Request request, ShardId shardId) throws IOExc
             IndexService indexService = indicesService.indexServiceSafe(request.getShard().getIndex());
             IndexShard indexShard = indexService.getShard(request.getShard().id());
             final SeqNoStats seqNoStats =  indexShard.seqNoStats();
-            final long indexMetaDataVersion = clusterService.state().metaData().index(shardId.getIndex()).getVersion();
+            final long mappingVersion = clusterService.state().metaData().index(shardId.getIndex()).getMappingVersion();
 
             final Translog.Operation[] operations = getOperations(
                     indexShard,
                     seqNoStats.getGlobalCheckpoint(),
                     request.fromSeqNo,
                     request.maxOperationCount,
                     request.maxOperationSizeInBytes);
-            return new Response(indexMetaDataVersion, seqNoStats.getGlobalCheckpoint(), seqNoStats.getMaxSeqNo(), operations);
+            return new Response(mappingVersion, seqNoStats.getGlobalCheckpoint(), seqNoStats.getMaxSeqNo(), operations);
         }
 
         @Override

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTask.java

@@ -81,7 +81,7 @@ public abstract class ShardFollowNodeTask extends AllocatedPersistentTask {
     private long followerMaxSeqNo = 0;
     private int numConcurrentReads = 0;
     private int numConcurrentWrites = 0;
-    private long currentIndexMetadataVersion = 0;
+    private long currentMappingVersion = 0;
     private long totalFetchTimeMillis = 0;
     private long numberOfSuccessfulFetches = 0;
     private long numberOfFailedFetches = 0;
@@ -139,14 +139,13 @@ void start(
             this.lastRequestedSeqNo = followerGlobalCheckpoint;
         }
 
-        // Forcefully updates follower mapping, this gets us the leader imd version and
-        // makes sure that leader and follower mapping are identical.
-        updateMapping(imdVersion -> {
+        // updates follower mapping, this gets us the leader mapping version and makes sure that leader and follower mapping are identical
+        updateMapping(mappingVersion -> {
             synchronized (ShardFollowNodeTask.this) {
-                currentIndexMetadataVersion = imdVersion;
+                currentMappingVersion = mappingVersion;
             }
-            LOGGER.info("{} Started to follow leader shard {}, followGlobalCheckPoint={}, indexMetaDataVersion={}",
-                params.getFollowShardId(), params.getLeaderShardId(), followerGlobalCheckpoint, imdVersion);
+            LOGGER.info("{} Started to follow leader shard {}, followGlobalCheckPoint={}, mappingVersion={}",
+                params.getFollowShardId(), params.getLeaderShardId(), followerGlobalCheckpoint, mappingVersion);
             coordinateReads();
         });
     }
@@ -269,7 +268,7 @@ private void sendShardChangesRequest(long from, int maxOperationCount, long maxR
     }
 
     void handleReadResponse(long from, long maxRequiredSeqNo, ShardChangesAction.Response response) {
-        maybeUpdateMapping(response.getIndexMetadataVersion(), () -> innerHandleReadResponse(from, maxRequiredSeqNo, response));
+        maybeUpdateMapping(response.getMappingVersion(), () -> innerHandleReadResponse(from, maxRequiredSeqNo, response));
     }
 
     /** Called when some operations are fetched from the leading */
@@ -355,16 +354,16 @@ private synchronized void handleWriteResponse(final BulkShardOperationsResponse
         coordinateReads();
     }
 
-    private synchronized void maybeUpdateMapping(Long minimumRequiredIndexMetadataVersion, Runnable task) {
-        if (currentIndexMetadataVersion >= minimumRequiredIndexMetadataVersion) {
-            LOGGER.trace("{} index metadata version [{}] is higher or equal than minimum required index metadata version [{}]",
-                params.getFollowShardId(), currentIndexMetadataVersion, minimumRequiredIndexMetadataVersion);
+    private synchronized void maybeUpdateMapping(Long minimumRequiredMappingVersion, Runnable task) {
+        if (currentMappingVersion >= minimumRequiredMappingVersion) {
+            LOGGER.trace("{} mapping version [{}] is higher or equal than minimum required mapping version [{}]",
+                params.getFollowShardId(), currentMappingVersion, minimumRequiredMappingVersion);
             task.run();
         } else {
-            LOGGER.trace("{} updating mapping, index metadata version [{}] is lower than minimum required index metadata version [{}]",
-                params.getFollowShardId(), currentIndexMetadataVersion, minimumRequiredIndexMetadataVersion);
-            updateMapping(imdVersion -> {
-                currentIndexMetadataVersion = imdVersion;
+            LOGGER.trace("{} updating mapping, mapping version [{}] is lower than minimum required mapping version [{}]",
+                params.getFollowShardId(), currentMappingVersion, minimumRequiredMappingVersion);
+            updateMapping(mappingVersion -> {
+                currentMappingVersion = mappingVersion;
                 task.run();
             });
         }
@@ -441,7 +440,7 @@ public synchronized Status getStatus() {
                 numConcurrentReads,
                 numConcurrentWrites,
                 buffer.size(),
-                currentIndexMetadataVersion,
+                currentMappingVersion,
                 totalFetchTimeMillis,
                 numberOfSuccessfulFetches,
                 numberOfFailedFetches,
@@ -469,7 +468,7 @@ public static class Status implements Task.Status {
         static final ParseField NUMBER_OF_CONCURRENT_READS_FIELD = new ParseField("number_of_concurrent_reads");
         static final ParseField NUMBER_OF_CONCURRENT_WRITES_FIELD = new ParseField("number_of_concurrent_writes");
         static final ParseField NUMBER_OF_QUEUED_WRITES_FIELD = new ParseField("number_of_queued_writes");
-        static final ParseField INDEX_METADATA_VERSION_FIELD = new ParseField("index_metadata_version");
+        static final ParseField MAPPING_VERSION_FIELD = new ParseField("mapping_version");
         static final ParseField TOTAL_FETCH_TIME_MILLIS_FIELD = new ParseField("total_fetch_time_millis");
         static final ParseField NUMBER_OF_SUCCESSFUL_FETCHES_FIELD = new ParseField("number_of_successful_fetches");
         static final ParseField NUMBER_OF_FAILED_FETCHES_FIELD = new ParseField("number_of_failed_fetches");
@@ -529,7 +528,7 @@ public static class Status implements Task.Status {
             STATUS_PARSER.declareInt(ConstructingObjectParser.constructorArg(), NUMBER_OF_CONCURRENT_READS_FIELD);
             STATUS_PARSER.declareInt(ConstructingObjectParser.constructorArg(), NUMBER_OF_CONCURRENT_WRITES_FIELD);
             STATUS_PARSER.declareInt(ConstructingObjectParser.constructorArg(), NUMBER_OF_QUEUED_WRITES_FIELD);
-            STATUS_PARSER.declareLong(ConstructingObjectParser.constructorArg(), INDEX_METADATA_VERSION_FIELD);
+            STATUS_PARSER.declareLong(ConstructingObjectParser.constructorArg(), MAPPING_VERSION_FIELD);
             STATUS_PARSER.declareLong(ConstructingObjectParser.constructorArg(), TOTAL_FETCH_TIME_MILLIS_FIELD);
             STATUS_PARSER.declareLong(ConstructingObjectParser.constructorArg(), NUMBER_OF_SUCCESSFUL_FETCHES_FIELD);
             STATUS_PARSER.declareLong(ConstructingObjectParser.constructorArg(), NUMBER_OF_FAILED_FETCHES_FIELD);
@@ -614,10 +613,10 @@ public int numberOfQueuedWrites() {
             return numberOfQueuedWrites;
         }
 
-        private final long indexMetadataVersion;
+        private final long mappingVersion;
 
-        public long indexMetadataVersion() {
-            return indexMetadataVersion;
+        public long mappingVersion() {
+            return mappingVersion;
         }
 
         private final long totalFetchTimeMillis;
@@ -697,7 +696,7 @@ public long timeSinceLastFetchMillis() {
                 final int numberOfConcurrentReads,
                 final int numberOfConcurrentWrites,
                 final int numberOfQueuedWrites,
-                final long indexMetadataVersion,
+                final long mappingVersion,
                 final long totalFetchTimeMillis,
                 final long numberOfSuccessfulFetches,
                 final long numberOfFailedFetches,
@@ -719,7 +718,7 @@ public long timeSinceLastFetchMillis() {
             this.numberOfConcurrentReads = numberOfConcurrentReads;
             this.numberOfConcurrentWrites = numberOfConcurrentWrites;
             this.numberOfQueuedWrites = numberOfQueuedWrites;
-            this.indexMetadataVersion = indexMetadataVersion;
+            this.mappingVersion = mappingVersion;
             this.totalFetchTimeMillis = totalFetchTimeMillis;
             this.numberOfSuccessfulFetches = numberOfSuccessfulFetches;
             this.numberOfFailedFetches = numberOfFailedFetches;
@@ -729,7 +728,7 @@ public long timeSinceLastFetchMillis() {
             this.numberOfSuccessfulBulkOperations = numberOfSuccessfulBulkOperations;
             this.numberOfFailedBulkOperations = numberOfFailedBulkOperations;
             this.numberOfOperationsIndexed = numberOfOperationsIndexed;
-            this.fetchExceptions = fetchExceptions;
+            this.fetchExceptions = Objects.requireNonNull(fetchExceptions);
             this.timeSinceLastFetchMillis = timeSinceLastFetchMillis;
         }
 
@@ -744,7 +743,7 @@ public Status(final StreamInput in) throws IOException {
             this.numberOfConcurrentReads = in.readVInt();
             this.numberOfConcurrentWrites = in.readVInt();
             this.numberOfQueuedWrites = in.readVInt();
-            this.indexMetadataVersion = in.readVLong();
+            this.mappingVersion = in.readVLong();
             this.totalFetchTimeMillis = in.readVLong();
             this.numberOfSuccessfulFetches = in.readVLong();
             this.numberOfFailedFetches = in.readVLong();
@@ -775,7 +774,7 @@ public void writeTo(final StreamOutput out) throws IOException {
             out.writeVInt(numberOfConcurrentReads);
             out.writeVInt(numberOfConcurrentWrites);
             out.writeVInt(numberOfQueuedWrites);
-            out.writeVLong(indexMetadataVersion);
+            out.writeVLong(mappingVersion);
             out.writeVLong(totalFetchTimeMillis);
             out.writeVLong(numberOfSuccessfulFetches);
             out.writeVLong(numberOfFailedFetches);
@@ -803,7 +802,7 @@ public XContentBuilder toXContent(final XContentBuilder builder, final Params pa
                 builder.field(NUMBER_OF_CONCURRENT_READS_FIELD.getPreferredName(), numberOfConcurrentReads);
                 builder.field(NUMBER_OF_CONCURRENT_WRITES_FIELD.getPreferredName(), numberOfConcurrentWrites);
                 builder.field(NUMBER_OF_QUEUED_WRITES_FIELD.getPreferredName(), numberOfQueuedWrites);
-                builder.field(INDEX_METADATA_VERSION_FIELD.getPreferredName(), indexMetadataVersion);
+                builder.field(MAPPING_VERSION_FIELD.getPreferredName(), mappingVersion);
                 builder.humanReadableField(
                         TOTAL_FETCH_TIME_MILLIS_FIELD.getPreferredName(),
                         "total_fetch_time",
@@ -867,14 +866,22 @@ public boolean equals(final Object o) {
                     numberOfConcurrentReads == that.numberOfConcurrentReads &&
                     numberOfConcurrentWrites == that.numberOfConcurrentWrites &&
                     numberOfQueuedWrites == that.numberOfQueuedWrites &&
-                    indexMetadataVersion == that.indexMetadataVersion &&
+                    mappingVersion == that.mappingVersion &&
                     totalFetchTimeMillis == that.totalFetchTimeMillis &&
                     numberOfSuccessfulFetches == that.numberOfSuccessfulFetches &&
                     numberOfFailedFetches == that.numberOfFailedFetches &&
                     operationsReceived == that.operationsReceived &&
                     totalTransferredBytes == that.totalTransferredBytes &&
                     numberOfSuccessfulBulkOperations == that.numberOfSuccessfulBulkOperations &&
                     numberOfFailedBulkOperations == that.numberOfFailedBulkOperations &&
+                    numberOfOperationsIndexed == that.numberOfOperationsIndexed &&
+                    /*
+                     * ElasticsearchException does not implement equals so we will assume the fetch exceptions are equal if they are equal
+                     * up to the key set and their messages.  Note that we are relying on the fact that the fetch exceptions are ordered by
+                     * keys.
+                     */
+                    fetchExceptions.keySet().equals(that.fetchExceptions.keySet()) &&
+                    getFetchExceptionMessages(this).equals(getFetchExceptionMessages(that)) &&
                     timeSinceLastFetchMillis == that.timeSinceLastFetchMillis;
         }
 
@@ -891,16 +898,26 @@ public int hashCode() {
                     numberOfConcurrentReads,
                     numberOfConcurrentWrites,
                     numberOfQueuedWrites,
-                    indexMetadataVersion,
+                    mappingVersion,
                     totalFetchTimeMillis,
                     numberOfSuccessfulFetches,
                     numberOfFailedFetches,
                     operationsReceived,
                     totalTransferredBytes,
                     numberOfSuccessfulBulkOperations,
                     numberOfFailedBulkOperations,
+                    numberOfOperationsIndexed,
+                    /*
+                     * ElasticsearchException does not implement hash code so we will compute the hash code based on the key set and the
+                     * messages. Note that we are relying on the fact that the fetch exceptions are ordered by keys.
+                     */
+                    fetchExceptions.keySet(),
+                    getFetchExceptionMessages(this),
                     timeSinceLastFetchMillis);
+        }
 
+        private static List<String> getFetchExceptionMessages(final Status status) {
+            return status.fetchExceptions().values().stream().map(ElasticsearchException::getMessage).collect(Collectors.toList());
         }
 
         public String toString() {

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTasksExecutor.java

@@ -115,7 +115,7 @@ protected void innerUpdateMapping(LongConsumer handler, Consumer<Exception> erro
                     putMappingRequest.type(mappingMetaData.type());
                     putMappingRequest.source(mappingMetaData.source().string(), XContentType.JSON);
                     followerClient.admin().indices().putMapping(putMappingRequest, ActionListener.wrap(
-                        putMappingResponse -> handler.accept(indexMetaData.getVersion()),
+                        putMappingResponse -> handler.accept(indexMetaData.getMappingVersion()),
                         errorHandler));
                 }, errorHandler));
             }

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/ShardChangesResponseTests.java

@@ -12,15 +12,15 @@ public class ShardChangesResponseTests extends AbstractStreamableTestCase<ShardC
 
     @Override
     protected ShardChangesAction.Response createTestInstance() {
-        final long indexMetadataVersion = randomNonNegativeLong();
+        final long mappingVersion = randomNonNegativeLong();
         final long leaderGlobalCheckpoint = randomNonNegativeLong();
         final long leaderMaxSeqNo = randomLongBetween(leaderGlobalCheckpoint, Long.MAX_VALUE);
         final int numOps = randomInt(8);
         final Translog.Operation[] operations = new Translog.Operation[numOps];
         for (int i = 0; i < numOps; i++) {
             operations[i] = new Translog.NoOp(i, 0, "test");
         }
-        return new ShardChangesAction.Response(indexMetadataVersion, leaderGlobalCheckpoint, leaderMaxSeqNo, operations);
+        return new ShardChangesAction.Response(mappingVersion, leaderGlobalCheckpoint, leaderMaxSeqNo, operations);
     }
 
     @Override