Add Sequence Numbers and enforce Primary Terms

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

@@ -0,0 +1,143 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.action;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+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 java.io.IOException;
+
+/**
+ * A base class for the response of a write operation that involves a single doc
+ */
+public abstract class DocWriteResponse extends ReplicationResponse implements ToXContent {
+
+    private ShardId shardId;
+    private String id;
+    private String type;
+    private long version;
+    private long seqNo;
+
+    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;
+    }
+
+    // needed for deserialization
+    protected DocWriteResponse() {
+    }
+
+    /**
+     * The index the document was changed in.
+     */
+    public String getIndex() {
+        return this.shardId.getIndex();
+    }
+
+
+    /**
+     * The exact shard the document was changed in.
+     */
+    public ShardId getShardId() {
+        return this.shardId;
+    }
+
+    /**
+     * The type of the document changed.
+     */
+    public String getType() {
+        return this.type;
+    }
+
+    /**
+     * The id of the document changed.
+     */
+    public String getId() {
+        return this.id;
+    }
+
+    /**
+     * Returns the current version of the doc.
+     */
+    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;
+    }
+
+
+    @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
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        shardId.writeTo(out);
+        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
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        ReplicationResponse.ShardInfo shardInfo = getShardInfo();
+        builder.field(Fields._INDEX, getIndex())
+                .field(Fields._TYPE, getType())
+                .field(Fields._ID, getId())
+                .field(Fields._VERSION, getVersion());
+        //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());
+        }
+        shardInfo.toXContent(builder, params);
+        return builder;
+    }
+}

core/src/main/java/org/elasticsearch/action/ActionWriteResponse.java → core/src/main/java/org/elasticsearch/action/ReplicationResponse.java

@@ -21,7 +21,6 @@
 
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ExceptionsHelper;
-import org.elasticsearch.bootstrap.Elasticsearch;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -30,25 +29,23 @@
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentBuilderString;
-import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.rest.RestStatus;
 
 import java.io.IOException;
-import java.util.Collections;
 
 /**
  * Base class for write action responses.
  */
-public class ActionWriteResponse extends ActionResponse {
+public class ReplicationResponse extends ActionResponse {
 
-    public final static ActionWriteResponse.ShardInfo.Failure[] EMPTY = new ActionWriteResponse.ShardInfo.Failure[0];
+    public final static ReplicationResponse.ShardInfo.Failure[] EMPTY = new ReplicationResponse.ShardInfo.Failure[0];
 
     private ShardInfo shardInfo;
 
     @Override
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        shardInfo = ActionWriteResponse.ShardInfo.readShardInfo(in);
+        shardInfo = ReplicationResponse.ShardInfo.readShardInfo(in);
     }
 
     @Override

core/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportFlushAction.java

@@ -19,7 +19,7 @@
 
 package org.elasticsearch.action.admin.indices.flush;
 
-import org.elasticsearch.action.ActionWriteResponse;
+import org.elasticsearch.action.ReplicationResponse;
 import org.elasticsearch.action.ShardOperationFailedException;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.replication.TransportBroadcastReplicationAction;
@@ -36,7 +36,7 @@
 /**
  * Flush Action.
  */
-public class TransportFlushAction extends TransportBroadcastReplicationAction<FlushRequest, FlushResponse, ShardFlushRequest, ActionWriteResponse> {
+public class TransportFlushAction extends TransportBroadcastReplicationAction<FlushRequest, FlushResponse, ShardFlushRequest, ReplicationResponse> {
 
     @Inject
     public TransportFlushAction(Settings settings, ThreadPool threadPool, ClusterService clusterService,
@@ -47,8 +47,8 @@ public TransportFlushAction(Settings settings, ThreadPool threadPool, ClusterSer
     }
 
     @Override
-    protected ActionWriteResponse newShardResponse() {
-        return new ActionWriteResponse();
+    protected ReplicationResponse newShardResponse() {
+        return new ReplicationResponse();
     }
 
     @Override

core/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportShardFlushAction.java

@@ -19,7 +19,7 @@
 
 package org.elasticsearch.action.admin.indices.flush;
 
-import org.elasticsearch.action.ActionWriteResponse;
+import org.elasticsearch.action.ReplicationResponse;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.replication.TransportReplicationAction;
 import org.elasticsearch.cluster.ClusterService;
@@ -42,7 +42,7 @@
 /**
  *
  */
-public class TransportShardFlushAction extends TransportReplicationAction<ShardFlushRequest, ShardFlushRequest, ActionWriteResponse> {
+public class TransportShardFlushAction extends TransportReplicationAction<ShardFlushRequest, ShardFlushRequest, ReplicationResponse> {
 
     public static final String NAME = FlushAction.NAME + "[s]";
 
@@ -56,16 +56,16 @@ public TransportShardFlushAction(Settings settings, TransportService transportSe
     }
 
     @Override
-    protected ActionWriteResponse newResponseInstance() {
-        return new ActionWriteResponse();
+    protected ReplicationResponse newResponseInstance() {
+        return new ReplicationResponse();
     }
 
     @Override
-    protected Tuple<ActionWriteResponse, ShardFlushRequest> shardOperationOnPrimary(ClusterState clusterState, PrimaryOperationRequest shardRequest) throws Throwable {
+    protected Tuple<ReplicationResponse, ShardFlushRequest> shardOperationOnPrimary(ClusterState clusterState, PrimaryOperationRequest shardRequest) throws Throwable {
         IndexShard indexShard = indicesService.indexServiceSafe(shardRequest.shardId.getIndex()).getShard(shardRequest.shardId.id());
         indexShard.flush(shardRequest.request.getRequest());
         logger.trace("{} flush request executed on primary", indexShard.shardId());
-        return new Tuple<>(new ActionWriteResponse(), shardRequest.request);
+        return new Tuple<>(new ReplicationResponse(), shardRequest.request);
     }
 
     @Override

core/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportRefreshAction.java

@@ -19,7 +19,7 @@
 
 package org.elasticsearch.action.admin.indices.refresh;
 
-import org.elasticsearch.action.ActionWriteResponse;
+import org.elasticsearch.action.ReplicationResponse;
 import org.elasticsearch.action.ShardOperationFailedException;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.replication.ReplicationRequest;
@@ -37,7 +37,7 @@
 /**
  * Refresh action.
  */
-public class TransportRefreshAction extends TransportBroadcastReplicationAction<RefreshRequest, RefreshResponse, ReplicationRequest, ActionWriteResponse> {
+public class TransportRefreshAction extends TransportBroadcastReplicationAction<RefreshRequest, RefreshResponse, ReplicationRequest, ReplicationResponse> {
 
     @Inject
     public TransportRefreshAction(Settings settings, ThreadPool threadPool, ClusterService clusterService,
@@ -48,8 +48,8 @@ public TransportRefreshAction(Settings settings, ThreadPool threadPool, ClusterS
     }
 
     @Override
-    protected ActionWriteResponse newShardResponse() {
-        return new ActionWriteResponse();
+    protected ReplicationResponse newShardResponse() {
+        return new ReplicationResponse();
     }
 
     @Override

core/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportShardRefreshAction.java

@@ -19,7 +19,7 @@
 
 package org.elasticsearch.action.admin.indices.refresh;
 
-import org.elasticsearch.action.ActionWriteResponse;
+import org.elasticsearch.action.ReplicationResponse;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.replication.ReplicationRequest;
 import org.elasticsearch.action.support.replication.TransportReplicationAction;
@@ -43,7 +43,7 @@
 /**
  *
  */
-public class TransportShardRefreshAction extends TransportReplicationAction<ReplicationRequest, ReplicationRequest, ActionWriteResponse> {
+public class TransportShardRefreshAction extends TransportReplicationAction<ReplicationRequest, ReplicationRequest, ReplicationResponse> {
 
     public static final String NAME = RefreshAction.NAME + "[s]";
 
@@ -57,16 +57,16 @@ public TransportShardRefreshAction(Settings settings, TransportService transport
     }
 
     @Override
-    protected ActionWriteResponse newResponseInstance() {
-        return new ActionWriteResponse();
+    protected ReplicationResponse newResponseInstance() {
+        return new ReplicationResponse();
     }
 
     @Override
-    protected Tuple<ActionWriteResponse, ReplicationRequest> shardOperationOnPrimary(ClusterState clusterState, PrimaryOperationRequest shardRequest) throws Throwable {
+    protected Tuple<ReplicationResponse, ReplicationRequest> shardOperationOnPrimary(ClusterState clusterState, PrimaryOperationRequest shardRequest) throws Throwable {
         IndexShard indexShard = indicesService.indexServiceSafe(shardRequest.shardId.getIndex()).getShard(shardRequest.shardId.id());
         indexShard.refresh("api");
         logger.trace("{} refresh request executed on primary", indexShard.shardId());
-        return new Tuple<>(new ActionWriteResponse(), shardRequest.request);
+        return new Tuple<>(new ReplicationResponse(), shardRequest.request);
     }
 
     @Override