Introduce Local checkpoints

core/src/main/java/org/elasticsearch/action/DocWriteResponse.java

@@ -21,9 +21,9 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.xcontent.StatusToXContent;
-import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentBuilderString;
+import org.elasticsearch.index.seqno.SequenceNumbersService;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.rest.RestStatus;
 
@@ -38,11 +38,13 @@ public abstract class DocWriteResponse extends ReplicationResponse implements St
     private String id;
     private String type;
     private long version;
+    private long seqNo;
 
-    public DocWriteResponse(ShardId shardId, String type, String id, long version) {
+    public DocWriteResponse(ShardId shardId, String type, String id, long seqNo, long version) {
         this.shardId = shardId;
         this.type = type;
         this.id = id;
+        this.seqNo = seqNo;
         this.version = version;
     }
 
@@ -86,19 +88,27 @@ public long getVersion() {
         return this.version;
     }
 
+    /**
+     * Returns the sequence number assigned for this change. Returns {@link SequenceNumbersService#UNASSIGNED_SEQ_NO} if the operation wasn't
+     * performed (i.e., an update operation that resulted in a NOOP).
+     */
+    public long getSeqNo() {
+        return seqNo;
+    }
+
     /** returns the rest status for this response (based on {@link ShardInfo#status()} */
     public RestStatus status() {
         return getShardInfo().status();
     }
 
-
     @Override
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         shardId = ShardId.readShardId(in);
         type = in.readString();
         id = in.readString();
         version = in.readZLong();
+        seqNo = in.readZLong();
     }
 
     @Override
@@ -108,13 +118,16 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeString(type);
         out.writeString(id);
         out.writeZLong(version);
+        out.writeZLong(seqNo);
     }
 
     static final class Fields {
         static final XContentBuilderString _INDEX = new XContentBuilderString("_index");
         static final XContentBuilderString _TYPE = new XContentBuilderString("_type");
         static final XContentBuilderString _ID = new XContentBuilderString("_id");
         static final XContentBuilderString _VERSION = new XContentBuilderString("_version");
+        static final XContentBuilderString _SHARD_ID = new XContentBuilderString("_shard_id");
+        static final XContentBuilderString _SEQ_NO = new XContentBuilderString("_seq_no");
     }
 
     @Override
@@ -125,6 +138,11 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             .field(Fields._ID, id)
             .field(Fields._VERSION, version);
         shardInfo.toXContent(builder, params);
+        //nocommit: i'm not sure we want to expose it in the api but it will be handy for debugging while we work...
+        builder.field(Fields._SHARD_ID, shardId.id());
+        if (getSeqNo() >= 0) {
+            builder.field(Fields._SEQ_NO, getSeqNo());
+        }
         return builder;
     }
 }

core/src/main/java/org/elasticsearch/action/admin/cluster/stats/TransportClusterStatsAction.java

@@ -19,7 +19,6 @@
 
 package org.elasticsearch.action.admin.cluster.stats;
 
-import org.elasticsearch.cluster.health.ClusterHealthStatus;
 import org.elasticsearch.action.admin.cluster.node.info.NodeInfo;
 import org.elasticsearch.action.admin.cluster.node.stats.NodeStats;
 import org.elasticsearch.action.admin.indices.stats.CommonStats;
@@ -30,6 +29,7 @@
 import org.elasticsearch.action.support.nodes.TransportNodesAction;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterService;
+import org.elasticsearch.cluster.health.ClusterHealthStatus;
 import org.elasticsearch.cluster.health.ClusterStateHealth;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.common.inject.Inject;
@@ -105,7 +105,8 @@ protected ClusterStatsNodeResponse nodeOperation(ClusterStatsNodeRequest nodeReq
             for (IndexShard indexShard : indexService) {
                 if (indexShard.routingEntry() != null && indexShard.routingEntry().active()) {
                     // only report on fully started shards
-                    shardsStats.add(new ShardStats(indexShard.routingEntry(), indexShard.shardPath(), new CommonStats(indexShard, SHARD_STATS_FLAGS), indexShard.commitStats()));
+                    shardsStats.add(new ShardStats(indexShard.routingEntry(), indexShard.shardPath(),
+                            new CommonStats(indexShard, SHARD_STATS_FLAGS), indexShard.commitStats(), indexShard.seqNoStats()));
                 }
             }
         }

core/src/main/java/org/elasticsearch/action/admin/indices/stats/CommonStatsFlags.java

@@ -227,7 +227,6 @@ public static enum Flag {
         RequestCache("request_cache"),
         Recovery("recovery");
 
-
         private final String restName;
 
         Flag(String restName) {

core/src/main/java/org/elasticsearch/action/admin/indices/stats/ShardStats.java

@@ -28,7 +28,7 @@
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentBuilderString;
 import org.elasticsearch.index.engine.CommitStats;
-import org.elasticsearch.index.shard.IndexShard;
+import org.elasticsearch.index.seqno.SeqNoStats;
 import org.elasticsearch.index.shard.ShardPath;
 
 import java.io.IOException;
@@ -42,20 +42,23 @@ public class ShardStats implements Streamable, ToXContent {
     private CommonStats commonStats;
     @Nullable
     private CommitStats commitStats;
+    @Nullable
+    private SeqNoStats seqNoStats;
     private String dataPath;
     private String statePath;
     private boolean isCustomDataPath;
 
     ShardStats() {
     }
 
-    public ShardStats(ShardRouting routing, ShardPath shardPath, CommonStats commonStats, CommitStats commitStats) {
+    public ShardStats(ShardRouting routing, ShardPath shardPath, CommonStats commonStats, CommitStats commitStats, SeqNoStats seqNoStats) {
         this.shardRouting = routing;
         this.dataPath = shardPath.getRootDataPath().toString();
         this.statePath = shardPath.getRootStatePath().toString();
         this.isCustomDataPath = shardPath.isCustomDataPath();
         this.commitStats = commitStats;
         this.commonStats = commonStats;
+        this.seqNoStats = seqNoStats;
     }
 
     /**
@@ -73,6 +76,11 @@ public CommitStats getCommitStats() {
         return this.commitStats;
     }
 
+    @Nullable
+    public SeqNoStats getSeqNoStats() {
+        return this.seqNoStats;
+    }
+
     public String getDataPath() {
         return dataPath;
     }
@@ -99,6 +107,7 @@ public void readFrom(StreamInput in) throws IOException {
         statePath = in.readString();
         dataPath = in.readString();
         isCustomDataPath = in.readBoolean();
+        seqNoStats = in.readOptionalStreamableReader(SeqNoStats::new);
     }
 
     @Override
@@ -109,6 +118,7 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeString(statePath);
         out.writeString(dataPath);
         out.writeBoolean(isCustomDataPath);
+        out.writeOptionalWritable(seqNoStats);
     }
 
     @Override
@@ -124,6 +134,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         if (commitStats != null) {
             commitStats.toXContent(builder, params);
         }
+        if (seqNoStats != null) {
+            seqNoStats.toXContent(builder, params);
+        }
         builder.startObject(Fields.SHARD_PATH);
         builder.field(Fields.STATE_PATH, statePath);
         builder.field(Fields.DATA_PATH, dataPath);

core/src/main/java/org/elasticsearch/action/admin/indices/stats/TransportIndicesStatsAction.java

@@ -162,6 +162,7 @@ protected ShardStats shardOperation(IndicesStatsRequest request, ShardRouting sh
             flags.set(CommonStatsFlags.Flag.Recovery);
         }
 
-        return new ShardStats(indexShard.routingEntry(), indexShard.shardPath(), new CommonStats(indexShard, flags), indexShard.commitStats());
+        return new ShardStats(indexShard.routingEntry(), indexShard.shardPath(),
+                new CommonStats(indexShard, flags), indexShard.commitStats(), indexShard.seqNoStats());
     }
 }

core/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

@@ -122,6 +122,9 @@ protected Tuple<BulkShardResponse, BulkShardRequest> shardOperationOnPrimary(Met
                     IndexResponse indexResponse = result.response();
                     setResponse(item, new BulkItemResponse(item.id(), indexRequest.opType().lowercase(), indexResponse));
                 } catch (Throwable e) {
+                    // nocommit: since we now have RetryOnPrimaryException, retrying doesn't always mean the shard is closed.
+                    // some operations were already perform and have a seqno assigned. we shouldn't just reindex them
+                    // if we have a pending mapping update
                     // rethrow the failure if we are going to retry on primary and let parent failure to handle it
                     if (retryPrimaryException(e)) {
                         // restore updated versions...
@@ -157,6 +160,8 @@ protected Tuple<BulkShardResponse, BulkShardRequest> shardOperationOnPrimary(Met
                     location = locationToSync(location, writeResult.location);
                     setResponse(item, new BulkItemResponse(item.id(), OP_TYPE_DELETE, deleteResponse));
                 } catch (Throwable e) {
+                    // nocommit: since we now have RetryOnPrimaryException, retrying doesn't always mean the shard is closed.
+                    // some operations were already perform and have a seqno assigned. we shouldn't just reindex them
                     // rethrow the failure if we are going to retry on primary and let parent failure to handle it
                     if (retryPrimaryException(e)) {
                         // restore updated versions...
@@ -204,7 +209,8 @@ protected Tuple<BulkShardResponse, BulkShardRequest> shardOperationOnPrimary(Met
                                 BytesReference indexSourceAsBytes = indexRequest.source();
                                 // add the response
                                 IndexResponse indexResponse = result.response();
-                                UpdateResponse updateResponse = new UpdateResponse(indexResponse.getShardInfo(), indexResponse.getShardId(), indexResponse.getType(), indexResponse.getId(), indexResponse.getVersion(), indexResponse.isCreated());
+                                UpdateResponse updateResponse = new UpdateResponse(indexResponse.getShardInfo(), indexResponse.getShardId(),
+                                    indexResponse.getType(), indexResponse.getId(), indexResponse.getSeqNo(),indexResponse.getVersion(), indexResponse.isCreated());
                                 if (updateRequest.fields() != null && updateRequest.fields().length > 0) {
                                     Tuple<XContentType, Map<String, Object>> sourceAndContent = XContentHelper.convertToMap(indexSourceAsBytes, true);
                                     updateResponse.setGetResult(updateHelper.extractGetResult(updateRequest, request.index(), indexResponse.getVersion(), sourceAndContent.v2(), sourceAndContent.v1(), indexSourceAsBytes));
@@ -216,7 +222,8 @@ protected Tuple<BulkShardResponse, BulkShardRequest> shardOperationOnPrimary(Met
                                 WriteResult<DeleteResponse> writeResult = updateResult.writeResult;
                                 DeleteResponse response = writeResult.response();
                                 DeleteRequest deleteRequest = updateResult.request();
-                                updateResponse = new UpdateResponse(response.getShardInfo(), response.getShardId(), response.getType(), response.getId(), response.getVersion(), false);
+                                updateResponse = new UpdateResponse(response.getShardInfo(), response.getShardId(), response.getType(),
+                                    response.getId(), response.getSeqNo(), response.getVersion(), false);
                                 updateResponse.setGetResult(updateHelper.extractGetResult(updateRequest, request.index(), response.getVersion(), updateResult.result.updatedSourceAsMap(), updateResult.result.updateSourceContentType(), null));
                                 // Replace the update request to the translated delete request to execute on the replica.
                                 item = request.items()[requestIndex] = new BulkItemRequest(request.items()[requestIndex].id(), deleteRequest);
@@ -238,6 +245,8 @@ protected Tuple<BulkShardResponse, BulkShardRequest> shardOperationOnPrimary(Met
                                         new BulkItemResponse.Failure(request.index(), updateRequest.type(), updateRequest.id(), t)));
                             }
                         } else {
+                            // nocommit: since we now have RetryOnPrimaryException, retrying doesn't always mean the shard is closed.
+                            // some operations were already perform and have a seqno assigned. we shouldn't just reindex them
                             // rethrow the failure if we are going to retry on primary and let parent failure to handle it
                             if (retryPrimaryException(t)) {
                                 // restore updated versions...
@@ -308,7 +317,7 @@ private void setResponse(BulkItemRequest request, BulkItemResponse response) {
         }
     }
 
-    private WriteResult shardIndexOperation(BulkShardRequest request, IndexRequest indexRequest, MetaData metaData,
+    private WriteResult<IndexResponse> shardIndexOperation(BulkShardRequest request, IndexRequest indexRequest, MetaData metaData,
                                             IndexShard indexShard, boolean processed) throws Throwable {
 
         // validate, if routing is required, that we got routing

core/src/main/java/org/elasticsearch/action/delete/DeleteResponse.java

@@ -43,12 +43,11 @@ public DeleteResponse() {
 
     }
 
-    public DeleteResponse(ShardId shardId, String type, String id, long version, boolean found) {
-        super(shardId, type, id, version);
+    public DeleteResponse(ShardId shardId, String type, String id, long seqNo, long version, boolean found) {
+        super(shardId, type, id, seqNo, version);
         this.found = found;
     }
 
-
     /**
      * Returns <tt>true</tt> if a doc was found to delete.
      */

core/src/main/java/org/elasticsearch/action/delete/TransportDeleteAction.java

@@ -137,15 +137,16 @@ public static WriteResult<DeleteResponse> executeDeleteRequestOnPrimary(DeleteRe
         // update the request with the version so it will go to the replicas
         request.versionType(delete.versionType().versionTypeForReplicationAndRecovery());
         request.version(delete.version());
+        request.seqNo(delete.seqNo());
 
         assert request.versionType().validateVersionForWrites(request.version());
         return new WriteResult<>(
-            new DeleteResponse(indexShard.shardId(), request.type(), request.id(), delete.version(), delete.found()),
+            new DeleteResponse(indexShard.shardId(), request.type(), request.id(), delete.seqNo(), delete.version(), delete.found()),
             delete.getTranslogLocation());
     }
 
     public static Engine.Delete executeDeleteRequestOnReplica(DeleteRequest request, IndexShard indexShard) {
-        Engine.Delete delete = indexShard.prepareDeleteOnReplica(request.type(), request.id(), request.version(), request.versionType());
+        Engine.Delete delete = indexShard.prepareDeleteOnReplica(request.type(), request.id(), request.seqNo(), request.version(), request.versionType());
         indexShard.delete(delete);
         return delete;
     }

core/src/main/java/org/elasticsearch/action/index/IndexResponse.java

@@ -40,11 +40,10 @@ public class IndexResponse extends DocWriteResponse {
     private boolean created;
 
     public IndexResponse() {
-
     }
 
-    public IndexResponse(ShardId shardId, String type, String id, long version, boolean created) {
-        super(shardId, type, id, version);
+    public IndexResponse(ShardId shardId, String type, String id, long seqNo, long version, boolean created) {
+        super(shardId, type, id, seqNo, version);
         this.created = created;
     }
 
@@ -84,6 +83,7 @@ public String toString() {
         builder.append(",id=").append(getId());
         builder.append(",version=").append(getVersion());
         builder.append(",created=").append(created);
+        builder.append(",seqNo=").append(getSeqNo());
         builder.append(",shards=").append(getShardInfo());
         return builder.append("]").toString();
     }

core/src/main/java/org/elasticsearch/action/index/TransportIndexAction.java

@@ -179,7 +179,7 @@ public static Engine.Index executeIndexRequestOnReplica(IndexRequest request, In
         SourceToParse sourceToParse = SourceToParse.source(SourceToParse.Origin.REPLICA, request.source()).index(shardId.getIndex()).type(request.type()).id(request.id())
                 .routing(request.routing()).parent(request.parent()).timestamp(request.timestamp()).ttl(request.ttl());
 
-        final Engine.Index operation = indexShard.prepareIndexOnReplica(sourceToParse, request.version(), request.versionType());
+        final Engine.Index operation = indexShard.prepareIndexOnReplica(sourceToParse, request.seqNo(), request.version(), request.versionType());
         Mapping update = operation.parsedDoc().dynamicMappingsUpdate();
         if (update != null) {
             throw new RetryOnReplicaException(shardId, "Mappings are not available on the replica yet, triggered update: " + update);
@@ -219,11 +219,11 @@ public static WriteResult<IndexResponse> executeIndexRequestOnPrimary(IndexReque
         final long version = operation.version();
         request.version(version);
         request.versionType(request.versionType().versionTypeForReplicationAndRecovery());
+        request.seqNo(operation.seqNo());
 
         assert request.versionType().validateVersionForWrites(request.version());
 
-        return new WriteResult<>(new IndexResponse(shardId, request.type(), request.id(), request.version(), created), operation.getTranslogLocation());
+        return new WriteResult<>(new IndexResponse(shardId, request.type(), request.id(), request.seqNo(), request.version(), created), operation.getTranslogLocation());
     }
-
 }