Add primary term to doc write response

This commit adds the primary term to the doc write response.

Author: jasontedor

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

@@ -57,6 +57,7 @@ public abstract class DocWriteResponse extends ReplicationResponse implements Wr
     private static final String _ID = "_id";
     private static final String _VERSION = "_version";
     private static final String _SEQ_NO = "_seq_no";
+    private static final String _PRIMARY_TERM = "_primary_term";
     private static final String RESULT = "result";
     private static final String FORCED_REFRESH = "forced_refresh";
 
@@ -116,14 +117,16 @@ public void writeTo(StreamOutput out) throws IOException {
     private String type;
     private long version;
     private long seqNo;
+    private long primaryTerm;
     private boolean forcedRefresh;
     protected Result result;
 
-    public DocWriteResponse(ShardId shardId, String type, String id, long seqNo, long version, Result result) {
+    public DocWriteResponse(ShardId shardId, String type, String id, long seqNo, long primaryTerm, long version, Result result) {
         this.shardId = shardId;
         this.type = type;
         this.id = id;
         this.seqNo = seqNo;
+        this.primaryTerm = primaryTerm;
         this.version = version;
         this.result = result;
     }
@@ -182,6 +185,15 @@ public long getSeqNo() {
         return seqNo;
     }
 
+    /**
+     * The primary term for this change.
+     *
+     * @return the primary term
+     */
+    public long getPrimaryTerm() {
+        return primaryTerm;
+    }
+
     /**
      * Did this request force a refresh? Requests that set {@link WriteRequest#setRefreshPolicy(RefreshPolicy)} to
      * {@link RefreshPolicy#IMMEDIATE} will always return true for this. Requests that set it to {@link RefreshPolicy#WAIT_UNTIL} will
@@ -251,8 +263,10 @@ public void readFrom(StreamInput in) throws IOException {
         version = in.readZLong();
         if (in.getVersion().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED)) {
             seqNo = in.readZLong();
+            primaryTerm = in.readVLong();
         } else {
             seqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
+            primaryTerm = 0;
         }
         forcedRefresh = in.readBoolean();
         result = Result.readFrom(in);
@@ -267,6 +281,7 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeZLong(version);
         if (out.getVersion().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED)) {
             out.writeZLong(seqNo);
+            out.writeVLong(primaryTerm);
         }
         out.writeBoolean(forcedRefresh);
         result.writeTo(out);
@@ -293,6 +308,7 @@ public XContentBuilder innerToXContent(XContentBuilder builder, Params params) t
         builder.field(_SHARDS, shardInfo);
         if (getSeqNo() >= 0) {
             builder.field(_SEQ_NO, getSeqNo());
+            builder.field(_PRIMARY_TERM, getPrimaryTerm());
         }
         return builder;
     }
@@ -333,7 +349,9 @@ protected static void parseInnerToXContent(XContentParser parser, Builder contex
                 context.setForcedRefresh(parser.booleanValue());
             } else if (_SEQ_NO.equals(currentFieldName)) {
                 context.setSeqNo(parser.longValue());
-            } else {
+            } else if (_PRIMARY_TERM.equals(currentFieldName)) {
+                context.setPrimaryTerm(parser.longValue());
+            } else{
                 throwUnknownField(currentFieldName, parser.getTokenLocation());
             }
         } else if (token == XContentParser.Token.START_OBJECT) {
@@ -362,6 +380,7 @@ public abstract static class Builder {
         protected boolean forcedRefresh;
         protected ShardInfo shardInfo = null;
         protected Long seqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
+        protected Long primaryTerm = 0L;
 
         public ShardId getShardId() {
             return shardId;
@@ -407,6 +426,10 @@ public void setSeqNo(Long seqNo) {
             this.seqNo = seqNo;
         }
 
+        public void setPrimaryTerm(Long primaryTerm) {
+            this.primaryTerm = primaryTerm;
+        }
+
         public abstract DocWriteResponse build();
     }
 }

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

@@ -139,7 +139,7 @@ private static BulkItemResultHolder executeIndexRequest(final IndexRequest index
             return new BulkItemResultHolder(null, indexResult, bulkItemRequest);
         } else {
             IndexResponse response = new IndexResponse(primary.shardId(), indexRequest.type(), indexRequest.id(),
-                    indexResult.getSeqNo(), indexResult.getVersion(), indexResult.isCreated());
+                    indexResult.getSeqNo(), indexResult.getPrimaryTerm(), indexResult.getVersion(), indexResult.isCreated());
             return new BulkItemResultHolder(response, indexResult, bulkItemRequest);
         }
     }
@@ -152,7 +152,7 @@ private static BulkItemResultHolder executeDeleteRequest(final DeleteRequest del
             return new BulkItemResultHolder(null, deleteResult, bulkItemRequest);
         } else {
             DeleteResponse response = new DeleteResponse(primary.shardId(), deleteRequest.type(), deleteRequest.id(),
-                    deleteResult.getSeqNo(), deleteResult.getVersion(), deleteResult.isFound());
+                    deleteResult.getSeqNo(), deleteResult.getPrimaryTerm(), deleteResult.getVersion(), deleteResult.isFound());
             return new BulkItemResultHolder(response, deleteResult, bulkItemRequest);
         }
     }
@@ -272,7 +272,7 @@ private static BulkItemResultHolder executeUpdateRequest(UpdateRequest updateReq
                                                              int requestIndex, UpdateHelper updateHelper,
                                                              LongSupplier nowInMillis,
                                                              final MappingUpdatePerformer mappingUpdater) throws Exception {
-        Engine.Result updateOperationResult = null;
+        Engine.Result result = null;
         UpdateResponse updateResponse = null;
         BulkItemRequest replicaRequest = request.items()[requestIndex];
         int maxAttempts = updateRequest.retryOnConflict();
@@ -284,7 +284,7 @@ private static BulkItemResultHolder executeUpdateRequest(UpdateRequest updateReq
             } catch (Exception failure) {
                 // we may fail translating a update to index or delete operation
                 // we use index result to communicate failure while translating update request
-                updateOperationResult = new Engine.IndexResult(failure, updateRequest.version(), SequenceNumbersService.UNASSIGNED_SEQ_NO);
+                result = new Engine.IndexResult(failure, updateRequest.version(), SequenceNumbersService.UNASSIGNED_SEQ_NO);
                 break; // out of retry loop
             }
             // execute translated update request
@@ -294,34 +294,41 @@ private static BulkItemResultHolder executeUpdateRequest(UpdateRequest updateReq
                     IndexRequest indexRequest = translate.action();
                     MappingMetaData mappingMd = metaData.mappingOrDefault(indexRequest.type());
                     indexRequest.process(mappingMd, request.index());
-                    updateOperationResult = executeIndexRequestOnPrimary(indexRequest, primary, mappingUpdater);
+                    result = executeIndexRequestOnPrimary(indexRequest, primary, mappingUpdater);
                     break;
                 case DELETED:
                     DeleteRequest deleteRequest = translate.action();
-                    updateOperationResult = executeDeleteRequestOnPrimary(deleteRequest, primary);
+                    result = executeDeleteRequestOnPrimary(deleteRequest, primary);
                     break;
                 case NOOP:
                     primary.noopUpdate(updateRequest.type());
                     break;
                 default: throw new IllegalStateException("Illegal update operation " + translate.getResponseResult());
             }
-            if (updateOperationResult == null) {
+            if (result == null) {
                 // this is a noop operation
                 updateResponse = translate.action();
                 break; // out of retry loop
-            } else if (updateOperationResult.hasFailure() == false) {
+            } else if (result.hasFailure() == false) {
                 // enrich update response and
                 // set translated update (index/delete) request for replica execution in bulk items
-                switch (updateOperationResult.getOperationType()) {
+                switch (result.getOperationType()) {
                     case INDEX:
+                        assert result instanceof Engine.IndexResult : result.getClass();
                         IndexRequest updateIndexRequest = translate.action();
-                        final IndexResponse indexResponse = new IndexResponse(primary.shardId(),
-                            updateIndexRequest.type(), updateIndexRequest.id(), updateOperationResult.getSeqNo(),
-                            updateOperationResult.getVersion(), ((Engine.IndexResult) updateOperationResult).isCreated());
+                        final IndexResponse indexResponse =
+                                new IndexResponse(primary.shardId(), updateIndexRequest.type(), updateIndexRequest.id(), result.getSeqNo(), result.getPrimaryTerm(),
+                            result.getVersion(), ((Engine.IndexResult) result).isCreated());
                         BytesReference indexSourceAsBytes = updateIndexRequest.source();
-                        updateResponse = new UpdateResponse(indexResponse.getShardInfo(),
-                            indexResponse.getShardId(), indexResponse.getType(), indexResponse.getId(), indexResponse.getSeqNo(),
-                            indexResponse.getVersion(), indexResponse.getResult());
+                        updateResponse = new UpdateResponse(
+                                indexResponse.getShardInfo(),
+                                indexResponse.getShardId(),
+                                indexResponse.getType(),
+                                indexResponse.getId(),
+                                indexResponse.getSeqNo(),
+                                indexResponse.getPrimaryTerm(),
+                                indexResponse.getVersion(),
+                                indexResponse.getResult());
                         if ((updateRequest.fetchSource() != null && updateRequest.fetchSource().fetchSource()) ||
                             (updateRequest.fields() != null && updateRequest.fields().length > 0)) {
                             Tuple<XContentType, Map<String, Object>> sourceAndContent =
@@ -333,29 +340,36 @@ private static BulkItemResultHolder executeUpdateRequest(UpdateRequest updateReq
                         replicaRequest = new BulkItemRequest(request.items()[requestIndex].id(), updateIndexRequest);
                         break;
                     case DELETE:
+                        assert result instanceof Engine.DeleteResult : result.getClass();
                         DeleteRequest updateDeleteRequest = translate.action();
                         DeleteResponse deleteResponse = new DeleteResponse(primary.shardId(),
-                            updateDeleteRequest.type(), updateDeleteRequest.id(), updateOperationResult.getSeqNo(),
-                            updateOperationResult.getVersion(), ((Engine.DeleteResult) updateOperationResult).isFound());
-                        updateResponse = new UpdateResponse(deleteResponse.getShardInfo(),
-                            deleteResponse.getShardId(), deleteResponse.getType(), deleteResponse.getId(), deleteResponse.getSeqNo(),
-                            deleteResponse.getVersion(), deleteResponse.getResult());
+                            updateDeleteRequest.type(), updateDeleteRequest.id(), result.getSeqNo(), result.getPrimaryTerm(),
+                            result.getVersion(), ((Engine.DeleteResult) result).isFound());
+                        updateResponse = new UpdateResponse(
+                                deleteResponse.getShardInfo(),
+                                deleteResponse.getShardId(),
+                                deleteResponse.getType(),
+                                deleteResponse.getId(),
+                                deleteResponse.getSeqNo(),
+                                deleteResponse.getPrimaryTerm(),
+                                deleteResponse.getVersion(),
+                                deleteResponse.getResult());
                         updateResponse.setGetResult(updateHelper.extractGetResult(updateRequest,
                             request.index(), deleteResponse.getVersion(), translate.updatedSourceAsMap(),
                             translate.updateSourceContentType(), null));
                         // set translated request as replica request
                         replicaRequest = new BulkItemRequest(request.items()[requestIndex].id(), updateDeleteRequest);
                         break;
                 }
-                assert updateOperationResult.getSeqNo() != SequenceNumbersService.UNASSIGNED_SEQ_NO;
+                assert result.getSeqNo() != SequenceNumbersService.UNASSIGNED_SEQ_NO;
                 // successful operation
                 break; // out of retry loop
-            } else if (updateOperationResult.getFailure() instanceof VersionConflictEngineException == false) {
+            } else if (result.getFailure() instanceof VersionConflictEngineException == false) {
                 // not a version conflict exception
                 break; // out of retry loop
             }
         }
-        return new BulkItemResultHolder(updateResponse, updateOperationResult, replicaRequest);
+        return new BulkItemResultHolder(updateResponse, result, replicaRequest);
     }
 
     static boolean shouldExecuteReplicaItem(final BulkItemRequest request, final int index) {
@@ -496,7 +510,7 @@ public static Engine.IndexResult executeIndexRequestOnPrimary(IndexRequest reque
                 mappingUpdater.updateMappings(mappingUpdate, primary.shardId(), request.type());
             }
         } catch (MapperParsingException | IllegalArgumentException failure) {
-            return new Engine.IndexResult(failure, request.version());
+            return new Engine.IndexResult(failure, request.version(), primary.getPrimaryTerm());
         }
 
         // Verify that there are no more mappings that need to be applied. If there are failures, a
@@ -509,7 +523,7 @@ public static Engine.IndexResult executeIndexRequestOnPrimary(IndexRequest reque
             } catch (MapperParsingException | IllegalStateException e) {
                 // there was an error in parsing the document that was not because
                 // of pending mapping updates, so return a failure for the result
-                return new Engine.IndexResult(e, request.version());
+                return new Engine.IndexResult(e, request.version(), primary.getPrimaryTerm());
             }
         } else {
             // There was no mapping update, the operation is the same as the pre-update version.

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

@@ -42,8 +42,8 @@ public class DeleteResponse extends DocWriteResponse {
     public DeleteResponse() {
     }
 
-    public DeleteResponse(ShardId shardId, String type, String id, long seqNo, long version, boolean found) {
-        super(shardId, type, id, seqNo, version, found ? Result.DELETED : Result.NOT_FOUND);
+    public DeleteResponse(ShardId shardId, String type, String id, long seqNo, long primaryTerm, long version, boolean found) {
+        super(shardId, type, id, seqNo, primaryTerm, version, found ? Result.DELETED : Result.NOT_FOUND);
     }
 
     @Override
@@ -112,7 +112,7 @@ public void setFound(boolean found) {
 
         @Override
         public DeleteResponse build() {
-            DeleteResponse deleteResponse = new DeleteResponse(shardId, type, id, seqNo, version, found);
+            DeleteResponse deleteResponse = new DeleteResponse(shardId, type, id, seqNo, primaryTerm, version, found);
             deleteResponse.setForcedRefresh(forcedRefresh);
             if (shardInfo != null) {
                 deleteResponse.setShardInfo(shardInfo);

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

@@ -43,8 +43,8 @@ public class IndexResponse extends DocWriteResponse {
     public IndexResponse() {
     }
 
-    public IndexResponse(ShardId shardId, String type, String id, long seqNo, long version, boolean created) {
-        super(shardId, type, id, seqNo, version, created ? Result.CREATED : Result.UPDATED);
+    public IndexResponse(ShardId shardId, String type, String id, long seqNo, long primaryTerm, long version, boolean created) {
+        super(shardId, type, id, seqNo, primaryTerm, version, created ? Result.CREATED : Result.UPDATED);
     }
 
     @Override
@@ -62,6 +62,7 @@ public String toString() {
         builder.append(",version=").append(getVersion());
         builder.append(",result=").append(getResult().getLowercase());
         builder.append(",seqNo=").append(getSeqNo());
+        builder.append(",primaryTerm=").append(getPrimaryTerm());
         builder.append(",shards=").append(Strings.toString(getShardInfo()));
         return builder.append("]").toString();
     }
@@ -114,7 +115,7 @@ public void setCreated(boolean created) {
 
         @Override
         public IndexResponse build() {
-            IndexResponse indexResponse = new IndexResponse(shardId, type, id, seqNo, version, created);
+            IndexResponse indexResponse = new IndexResponse(shardId, type, id, seqNo, primaryTerm, version, created);
             indexResponse.setForcedRefresh(forcedRefresh);
             if (shardInfo != null) {
                 indexResponse.setShardInfo(shardInfo);