Add BWC layer to seq no infra and enable BWC tests

.gitignore

@@ -43,6 +43,5 @@ html_docs
 
 # random old stuff that we should look at the necessity of...
 /tmp/
-backwards/
 eclipse-build
 

buildSrc/src/main/groovy/org/elasticsearch/gradle/test/ClusterFormationTasks.groovy

@@ -268,6 +268,7 @@ class ClusterFormationTasks {
     static Task configureWriteConfigTask(String name, Project project, Task setup, NodeInfo node, NodeInfo seedNode) {
         Map esConfig = [
                 'cluster.name'                 : node.clusterName,
+                'node.name'                    : "node-" + node.nodeNum,
                 'pidfile'                      : node.pidFile,
                 'path.repo'                    : "${node.sharedDir}/repo",
                 'path.shared_data'             : "${node.sharedDir}/",

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

@@ -18,6 +18,7 @@
  */
 package org.elasticsearch.action;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.action.support.WriteRequest.RefreshPolicy;
 import org.elasticsearch.action.support.WriteResponse;
@@ -214,7 +215,11 @@ public void readFrom(StreamInput in) throws IOException {
         type = in.readString();
         id = in.readString();
         version = in.readZLong();
-        seqNo = in.readZLong();
+        if (in.getVersion().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED)) {
+            seqNo = in.readZLong();
+        } else {
+            seqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
+        }
         forcedRefresh = in.readBoolean();
         result = Result.readFrom(in);
     }
@@ -226,7 +231,9 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeString(type);
         out.writeString(id);
         out.writeZLong(version);
-        out.writeZLong(seqNo);
+        if (out.getVersion().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED)) {
+            out.writeZLong(seqNo);
+        }
         out.writeBoolean(forcedRefresh);
         result.writeTo(out);
     }

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

@@ -58,6 +58,6 @@ public void writeTo(StreamOutput out) throws IOException {
 
     @Override
     public String toString() {
-        return "flush {" + super.toString() + "}";
+        return "flush {" + shardId + "}";
     }
 }

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

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.action.admin.indices.stats;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -102,7 +103,9 @@ public void readFrom(StreamInput in) throws IOException {
         statePath = in.readString();
         dataPath = in.readString();
         isCustomDataPath = in.readBoolean();
-        seqNoStats = in.readOptionalWriteable(SeqNoStats::new);
+        if (in.getVersion().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED)) {
+            seqNoStats = in.readOptionalWriteable(SeqNoStats::new);
+        }
     }
 
     @Override
@@ -113,7 +116,9 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeString(statePath);
         out.writeString(dataPath);
         out.writeBoolean(isCustomDataPath);
-        out.writeOptionalWriteable(seqNoStats);
+        if (out.getVersion().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED)) {
+            out.writeOptionalWriteable(seqNoStats);
+        }
     }
 
     @Override

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

@@ -524,7 +524,9 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeOptionalString(routing);
         out.writeOptionalString(parent);
         if (out.getVersion().before(Version.V_6_0_0_alpha1_UNRELEASED)) {
-            out.writeOptionalString(null);
+            // timestamp, at this point #proccess was called which for previous versions meant this was set
+            // nocommit: can we fix this in 5.x? how?
+            out.writeOptionalString("0");
             out.writeOptionalWriteable(null);
         }
         out.writeBytesReference(source);

core/src/main/java/org/elasticsearch/action/support/replication/BasicReplicationRequest.java

@@ -37,4 +37,9 @@ public BasicReplicationRequest() {
     public BasicReplicationRequest(ShardId shardId) {
         super(shardId);
     }
+
+    @Override
+    public String toString() {
+        return "BasicReplicationRequest{" + shardId + "}";
+    }
 }

core/src/main/java/org/elasticsearch/action/support/replication/ReplicatedWriteRequest.java

@@ -19,12 +19,14 @@
 
 package org.elasticsearch.action.support.replication;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.bulk.BulkShardRequest;
 import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.index.seqno.SequenceNumbersService;
 import org.elasticsearch.index.shard.ShardId;
 
 import java.io.IOException;
@@ -36,6 +38,9 @@
 public abstract class ReplicatedWriteRequest<R extends ReplicatedWriteRequest<R>> extends ReplicationRequest<R> implements WriteRequest<R> {
     private RefreshPolicy refreshPolicy = RefreshPolicy.NONE;
 
+    long seqNo;
+
+
     /**
      * Constructor for deserialization.
      */
@@ -62,11 +67,32 @@ public RefreshPolicy getRefreshPolicy() {
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         refreshPolicy = RefreshPolicy.readFrom(in);
+        if (in.getVersion().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED))  {
+            seqNo = in.readVLong();
+        } else {
+            seqNo = SequenceNumbersService.UNASSIGNED_SEQ_NO;
+        }
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         refreshPolicy.writeTo(out);
+        if (out.getVersion().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED)) {
+            out.writeVLong(seqNo);
+        }
+    }
+
+    /**
+     * Returns the sequence number for this operation. The sequence number is assigned while the operation
+     * is performed on the primary shard.
+     */
+    public long seqNo() {
+        return seqNo;
+    }
+
+    /** sets the sequence number for this operation. should only be called on the primary shard */
+    public void seqNo(long seqNo) {
+        this.seqNo = seqNo;
     }
 }

core/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java

@@ -283,7 +283,7 @@ protected List<ShardRouting> getShards(ShardId shardId, ClusterState state) {
     }
 
     private void decPendingAndFinishIfNeeded() {
-        assert pendingActions.get() > 0;
+        assert pendingActions.get() > 0 : "pending action count goes bellow 0 for request [" + request + "]";
         if (pendingActions.decrementAndGet() == 0) {
             finish();
         }

core/src/main/java/org/elasticsearch/action/support/replication/ReplicationRequest.java

@@ -55,7 +55,6 @@ public abstract class ReplicationRequest<Request extends ReplicationRequest<Requ
      */
     protected ShardId shardId;
 
-    long seqNo;
     long primaryTerm;
 
     protected TimeValue timeout = DEFAULT_TIMEOUT;
@@ -171,19 +170,6 @@ long routedBasedOnClusterVersion() {
         return routedBasedOnClusterVersion;
     }
 
-    /**
-     * Returns the sequence number for this operation. The sequence number is assigned while the operation
-     * is performed on the primary shard.
-     */
-    public long seqNo() {
-        return seqNo;
-    }
-
-    /** sets the sequence number for this operation. should only be called on the primary shard */
-    public void seqNo(long seqNo) {
-        this.seqNo = seqNo;
-    }
-
     /** returns the primary term active at the time the operation was performed on the primary shard */
     public long primaryTerm() {
         return primaryTerm;
@@ -215,7 +201,6 @@ public void readFrom(StreamInput in) throws IOException {
         timeout = new TimeValue(in);
         index = in.readString();
         routedBasedOnClusterVersion = in.readVLong();
-        seqNo = in.readVLong();
         primaryTerm = in.readVLong();
     }
 
@@ -232,7 +217,6 @@ public void writeTo(StreamOutput out) throws IOException {
         timeout.writeTo(out);
         out.writeString(index);
         out.writeVLong(routedBasedOnClusterVersion);
-        out.writeVLong(seqNo);
         out.writeVLong(primaryTerm);
     }
 
@@ -252,13 +236,7 @@ public Request setShardId(ShardId shardId) {
     }
 
     @Override
-    public String toString() {
-        if (shardId != null) {
-            return shardId.toString();
-        } else {
-            return index;
-        }
-    }
+    public abstract String toString(); // force a proper to string to ease debugging
 
     @Override
     public String getDescription() {

core/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java

@@ -21,6 +21,7 @@
 
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionListenerResponseHandler;
 import org.elasticsearch.action.ActionResponse;
@@ -983,16 +984,26 @@ public ReplicaResponse(String allocationId, long localCheckpoint) {
 
         @Override
         public void readFrom(StreamInput in) throws IOException {
-            super.readFrom(in);
-            localCheckpoint = in.readZLong();
-            allocationId = in.readString();
+            if (in.getVersion().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED)) {
+                super.readFrom(in);
+                localCheckpoint = in.readZLong();
+                allocationId = in.readString();
+            } else {
+                // we use to read empty responses
+                Empty.INSTANCE.readFrom(in);
+            }
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
-            super.writeTo(out);
-            out.writeZLong(localCheckpoint);
-            out.writeString(allocationId);
+            if (out.getVersion().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED)) {
+                super.writeTo(out);
+                out.writeZLong(localCheckpoint);
+                out.writeString(allocationId);
+            } else {
+                // we use to write empty responses
+                Empty.INSTANCE.writeTo(out);
+            }
         }
 
         @Override
@@ -1016,10 +1027,9 @@ public void performOn(ShardRouting replica, ReplicaRequest request, ActionListen
                 listener.onFailure(new NoNodeAvailableException("unknown node [" + nodeId + "]"));
                 return;
             }
-            transportService.sendRequest(node, transportReplicaAction,
-                new ConcreteShardRequest<>(request, replica.allocationId().getId()), transportOptions,
-                // Eclipse can't handle when this is <> so we specify the type here.
-                new ActionListenerResponseHandler<ReplicaResponse>(listener, ReplicaResponse::new));
+            final ConcreteShardRequest<ReplicaRequest> concreteShardRequest =
+                new ConcreteShardRequest<>(request, replica.allocationId().getId());
+            sendReplicaRequest(concreteShardRequest, node, listener);
         }
 
         @Override
@@ -1060,6 +1070,14 @@ public void onFailure(Exception shardFailedError) {
         }
     }
 
+    /** sends the give replica request to the supplied nodes */
+    protected void sendReplicaRequest(ConcreteShardRequest<ReplicaRequest> concreteShardRequest, DiscoveryNode node,
+                                      ActionListener<ReplicationOperation.ReplicaResponse> listener) {
+        transportService.sendRequest(node, transportReplicaAction, concreteShardRequest, transportOptions,
+            // Eclipse can't handle when this is <> so we specify the type here.
+            new ActionListenerResponseHandler<>(listener, ReplicaResponse::new));
+    }
+
     /** a wrapper class to encapsulate a request when being sent to a specific allocation id **/
     public static final class ConcreteShardRequest<R extends TransportRequest> extends TransportRequest {
 

core/src/main/java/org/elasticsearch/cluster/metadata/MappingMetaData.java

@@ -204,7 +204,7 @@ public void writeTo(StreamOutput out) throws IOException {
             // timestamp
             out.writeBoolean(false); // enabled
             out.writeString(DateFieldMapper.DEFAULT_DATE_TIME_FORMATTER.format());
-            out.writeOptionalString(null);
+            out.writeOptionalString("now"); // old default
             out.writeOptionalBoolean(null);
         }
         out.writeBoolean(hasParentField());

core/src/main/java/org/elasticsearch/index/seqno/GlobalCheckpointSyncAction.java

@@ -20,29 +20,28 @@
 
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.logging.log4j.util.Supplier;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.replication.ReplicationOperation;
 import org.elasticsearch.action.support.replication.ReplicationRequest;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
 import org.elasticsearch.action.support.replication.TransportReplicationAction;
 import org.elasticsearch.cluster.action.shard.ShardStateAction;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
-import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
-import java.io.UncheckedIOException;
-import java.io.UnsupportedEncodingException;
 
 public class GlobalCheckpointSyncAction extends TransportReplicationAction<GlobalCheckpointSyncAction.PrimaryRequest,
     GlobalCheckpointSyncAction.ReplicaRequest, ReplicationResponse> {
@@ -65,6 +64,17 @@ protected ReplicationResponse newResponseInstance() {
         return new ReplicationResponse();
     }
 
+    @Override
+    protected void sendReplicaRequest(ConcreteShardRequest<ReplicaRequest> concreteShardRequest, DiscoveryNode node,
+                                      ActionListener<ReplicationOperation.ReplicaResponse> listener) {
+        if (node.getVersion().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED)) {
+            super.sendReplicaRequest(concreteShardRequest, node, listener);
+        } else {
+            listener.onResponse(
+                new ReplicaResponse(concreteShardRequest.getTargetAllocationID(), SequenceNumbersService.UNASSIGNED_SEQ_NO));
+        }
+    }
+
     @Override
     protected PrimaryResult shardOperationOnPrimary(PrimaryRequest request, IndexShard indexShard) throws Exception {
         long checkpoint = indexShard.getGlobalCheckpoint();
@@ -105,6 +115,11 @@ private PrimaryRequest() {
         public PrimaryRequest(ShardId shardId) {
             super(shardId);
         }
+
+        @Override
+        public String toString() {
+            return "GlobalCkpSyncPrimary{" + shardId + "}";
+        }
     }
 
     public static final class ReplicaRequest extends ReplicationRequest<GlobalCheckpointSyncAction.ReplicaRequest> {
@@ -134,6 +149,14 @@ public void writeTo(StreamOutput out) throws IOException {
         public long getCheckpoint() {
             return checkpoint;
         }
+
+        @Override
+        public String toString() {
+            return "GlobalCkpSyncReplica{" +
+                "checkpoint=" + checkpoint +
+                ", shardId=" + shardId +
+                '}';
+        }
     }
 
 }