Reshard add shard count #133985
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -9,6 +9,7 @@ | |
|
|
||
| package org.elasticsearch.action.support.replication; | ||
|
|
||
| import org.elasticsearch.TransportVersions; | ||
| import org.elasticsearch.action.ActionRequestValidationException; | ||
| import org.elasticsearch.action.IndicesRequest; | ||
| import org.elasticsearch.action.LegacyActionRequest; | ||
|
|
@@ -49,6 +50,19 @@ public abstract class ReplicationRequest<Request extends ReplicationRequest<Requ | |
| protected TimeValue timeout; | ||
| protected String index; | ||
|
|
||
| /** | ||
| * The reshardSplitShardCount has been added to accommodate the Resharding project. | ||
ankikuma marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| * This is populated when the coordinator is deciding which shards a request applies to. | ||
| * For example, {@link org.elasticsearch.action.bulk.BulkOperation} splits | ||
| * an incoming bulk request into shard level {@link org.elasticsearch.action.bulk.BulkShardRequest} | ||
| * based on its' cluster state view of the number of shards that are ready for indexing. | ||
ankikuma marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| * The purpose of this metadata is to reconcile the cluster state visible at the coordinating | ||
| * node with that visible at the source shard node. (w.r.t resharding). | ||
bcully marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| * Note that we are able to get away with a single number, instead of an array of target shard states, | ||
| * because we only allow splits in increments of 2x. | ||
|
There was a problem hiding this comment. I'm tempted to want concrete examples here. I know this makes for a long comment, but I think the field is unintuitive enough to warrant it. There was a problem hiding this comment. I agree that examples will be helpful. Its semantics really is confusing. Also, I was thinking if I rename the field in the There was a problem hiding this comment. I think we referred to this value as a "checksum of resharding state" in the design document. I wonder if calling it some kind of checksum will resolve the naming dispute. There was a problem hiding this comment. Ya I like reshardSplitShardCountChecksum. Because otherwise it sounds like it's a count that you can actually use for something. It's really just a value to be used not as is, but in the context of reconciling cluster state between 2 different nodes. There was a problem hiding this comment. bikeshedding, how about "summary" instead of "checksum"? There was a problem hiding this comment. I kind of like checksum better but happy change it ... There was a problem hiding this comment. Ok, I won't push hard on this. To me checksum implies an integrity check - you have some data, and alongside it you have a provided checksum, and if you recompute the checksum and it doesn't match the provided checksum you have an integrity problem. It sort of applies here - we have our own view of the routing table, and the coordinator provides a "checksum", and if we don't produce the same value over our view then we don't trust the coordinator's binning (which kind of looks like a response to an integrity error if you squint). I felt that summary (i.e., a lossily compressed form that still has relevant info) was maybe a little more accurate, but I fully admit I'm bikeshedding. Carry on as you see fit. |
||
| */ | ||
| protected final int reshardSplitShardCount; | ||
|
|
||
| /** | ||
| * The number of shard copies that must be active before proceeding with the replication action. | ||
| */ | ||
|
|
@@ -61,6 +75,10 @@ public ReplicationRequest(StreamInput in) throws IOException { | |
| } | ||
|
|
||
| public ReplicationRequest(@Nullable ShardId shardId, StreamInput in) throws IOException { | ||
| this(shardId, 0, in); | ||
| } | ||
|
|
||
| public ReplicationRequest(@Nullable ShardId shardId, int reshardSplitShardCount, StreamInput in) throws IOException { | ||
|
||
| super(in); | ||
| final boolean thinRead = shardId != null; | ||
| if (thinRead) { | ||
|
|
@@ -80,15 +98,28 @@ public ReplicationRequest(@Nullable ShardId shardId, StreamInput in) throws IOEx | |
| index = in.readString(); | ||
| } | ||
| routedBasedOnClusterVersion = in.readVLong(); | ||
| if (in.getTransportVersion().onOrAfter(TransportVersions.INDEX_RESHARD_SHARDCOUNT_REPLICATION_REQUEST) && (thinRead == false)) { | ||
| this.reshardSplitShardCount = in.readInt(); | ||
| } else { | ||
| this.reshardSplitShardCount = reshardSplitShardCount; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Creates a new request with resolved shard id | ||
| */ | ||
| public ReplicationRequest(@Nullable ShardId shardId) { | ||
| this(shardId, 0); | ||
| } | ||
|
|
||
| /** | ||
| * Creates a new request with resolved shard id and reshardSplitShardCount | ||
| */ | ||
| public ReplicationRequest(@Nullable ShardId shardId, int reshardSplitShardCount) { | ||
| this.index = shardId == null ? null : shardId.getIndexName(); | ||
| this.shardId = shardId; | ||
| this.timeout = DEFAULT_TIMEOUT; | ||
| this.reshardSplitShardCount = reshardSplitShardCount; | ||
| } | ||
|
|
||
| /** | ||
|
|
@@ -137,6 +168,14 @@ public ShardId shardId() { | |
| return shardId; | ||
| } | ||
|
|
||
| /** | ||
| * @return The effective shard count as seen by the coordinator when creating this request. | ||
| * can be 0 if this has not yet been resolved. | ||
| */ | ||
| public int reshardSplitShardCount() { | ||
| return reshardSplitShardCount; | ||
| } | ||
|
|
||
| /** | ||
| * Sets the number of shard copies that must be active before proceeding with the replication | ||
| * operation. Defaults to {@link ActiveShardCount#DEFAULT}, which requires one shard copy | ||
|
|
@@ -191,11 +230,14 @@ public void writeTo(StreamOutput out) throws IOException { | |
| out.writeTimeValue(timeout); | ||
| out.writeString(index); | ||
| out.writeVLong(routedBasedOnClusterVersion); | ||
| if (out.getTransportVersion().onOrAfter(TransportVersions.INDEX_RESHARD_SHARDCOUNT_REPLICATION_REQUEST)) { | ||
| out.writeInt(reshardSplitShardCount); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Thin serialization that does not write {@link #shardId} and will only write {@link #index} if it is different from the index name in | ||
| * {@link #shardId}. Since we do not write {@link #shardId}, we also do not write {@link #reshardSplitShardCount}. | ||
| */ | ||
| public void writeThin(StreamOutput out) throws IOException { | ||
| super.writeTo(out); | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.