Skip to content

Commit

Permalink
Only fetch mapping updates when necessary (#33182)
Browse files Browse the repository at this point in the history 
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.
  • Loading branch information
@jasontedor
jasontedor committed Aug 28, 2018
1 parent 36feede commit c5b6ccc
Show file tree
Hide file tree
Showing 9 changed files with 87 additions and 88 deletions.
22 changes: 11 additions & 11 deletions x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardChangesAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -168,10 +168,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;
Expand All @@ -195,8 +195,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;
Expand All @@ -205,7 +205,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);
Expand All @@ -214,7 +214,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);
Expand All @@ -225,15 +225,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));
}
}

Expand Down Expand Up @@ -266,15 +266,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
Expand Down
57 changes: 28 additions & 29 deletions x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTask.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
});
}
Expand Down Expand Up @@ -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 */
Expand Down Expand Up @@ -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();
});
}
Expand Down Expand Up @@ -422,7 +421,7 @@ public synchronized Status getStatus() {
numConcurrentReads,
numConcurrentWrites,
buffer.size(),
currentIndexMetadataVersion,
currentMappingVersion,
totalFetchTimeMillis,
numberOfSuccessfulFetches,
numberOfFailedFetches,
Expand All @@ -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");
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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,
Expand All @@ -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;
Expand All @@ -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();
Expand Down Expand Up @@ -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);
Expand All @@ -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",
Expand Down Expand Up @@ -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 &&
Expand All @@ -837,7 +836,7 @@ public int hashCode() {
numberOfConcurrentReads,
numberOfConcurrentWrites,
numberOfQueuedWrites,
indexMetadataVersion,
mappingVersion,
totalFetchTimeMillis,
numberOfSuccessfulFetches,
numberOfFailedFetches,
Expand Down
2 changes: 1 addition & 1 deletion ...plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTasksExecutor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -116,7 +116,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));
}
Expand Down
4 changes: 2 additions & 2 deletions ...lugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/ShardChangesResponseTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading

0 comments on commit c5b6ccc

Please sign in to comment.