Only fetch mapping updates when necessary (#33182)

Today we fetch the mapping from the leader and apply it as a mapping
update whenever the index metadata version on the leader changes. Yet,
the index metadata can change for many reasons other than a mapping
update (e.g., settings updates, adding an alias, or a replica being
promoted to a primary among many other reasons). This commit builds on
the addition of a mapping version to the index metadata to only fetch
mapping updates when the mapping version increases. This reduces the
number of these fetches and application of mappings on the follower to
the bare minimum.

Author: jasontedor

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

@@ -80,7 +80,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;
@@ -131,14 +131,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();
         });
     }
@@ -258,7 +257,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 */
@@ -344,16 +343,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();
             });
         }
@@ -422,7 +421,7 @@ public synchronized Status getStatus() {
                 numConcurrentReads,
                 numConcurrentWrites,
                 buffer.size(),
-                currentIndexMetadataVersion,
+                currentMappingVersion,
                 totalFetchTimeMillis,
                 numberOfSuccessfulFetches,
                 numberOfFailedFetches,
@@ -448,7 +447,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");
@@ -504,7 +503,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);
@@ -582,10 +581,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;
@@ -658,7 +657,7 @@ public NavigableMap<Long, ElasticsearchException> fetchExceptions() {
                 final int numberOfConcurrentReads,
                 final int numberOfConcurrentWrites,
                 final int numberOfQueuedWrites,
-                final long indexMetadataVersion,
+                final long mappingVersion,
                 final long totalFetchTimeMillis,
                 final long numberOfSuccessfulFetches,
                 final long numberOfFailedFetches,
@@ -678,7 +677,7 @@ public NavigableMap<Long, ElasticsearchException> fetchExceptions() {
             this.numberOfConcurrentReads = numberOfConcurrentReads;
             this.numberOfConcurrentWrites = numberOfConcurrentWrites;
             this.numberOfQueuedWrites = numberOfQueuedWrites;
-            this.indexMetadataVersion = indexMetadataVersion;
+            this.mappingVersion = mappingVersion;
             this.totalFetchTimeMillis = totalFetchTimeMillis;
             this.numberOfSuccessfulFetches = numberOfSuccessfulFetches;
             this.numberOfFailedFetches = numberOfFailedFetches;
@@ -701,7 +700,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();
@@ -730,7 +729,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);
@@ -756,7 +755,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",
@@ -815,7 +814,7 @@ 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 &&
@@ -837,7 +836,7 @@ public int hashCode() {
                     numberOfConcurrentReads,
                     numberOfConcurrentWrites,
                     numberOfQueuedWrites,
-                    indexMetadataVersion,
+                    mappingVersion,
                     totalFetchTimeMillis,
                     numberOfSuccessfulFetches,
                     numberOfFailedFetches,

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