From f121cd3beb4b1b5a3acd991a099bbe4177eaa547 Mon Sep 17 00:00:00 2001
From: Jason Tedor <jason@tedor.me>
Date: Mon, 17 Jul 2017 17:47:48 +0900
Subject: [PATCH] Fix pre-6.0 response to unknown replication actions

When sending replica requests for replication operations, we skip
sending the request to pre-6.0 nodes for operations that such nodes
would not be aware of (e.g., the background global checkpoint sync, or
the primary/replica resync) since they would not know what to do with
these requests. Yet, we simulate that we received responses from these
nodes. Today, this is done by simulating that they sent us that their
local checkpoint is unassigned sequence number. However, for pre-6.0
nodes we have introduced a special local checkpoint used in the global
checkpoint tracker for such nodes and that is what we should use here
too. This commit fixes this issue.

Relates #25744
---
 .../action/resync/TransportResyncReplicationAction.java     | 2 +-
 .../support/replication/TransportReplicationAction.java     | 6 ++++++
 .../index/seqno/GlobalCheckpointSyncAction.java             | 2 +-
 .../elasticsearch/index/seqno/GlobalCheckpointTracker.java  | 3 +++
 .../index/seqno/GlobalCheckpointTrackerTests.java           | 2 +-
 5 files changed, 12 insertions(+), 3 deletions(-)

diff --git a/core/src/main/java/org/elasticsearch/action/resync/TransportResyncReplicationAction.java b/core/src/main/java/org/elasticsearch/action/resync/TransportResyncReplicationAction.java
index c723a175ad7..6b84522d23a 100644
--- a/core/src/main/java/org/elasticsearch/action/resync/TransportResyncReplicationAction.java
+++ b/core/src/main/java/org/elasticsearch/action/resync/TransportResyncReplicationAction.java
@@ -93,7 +93,7 @@ public class TransportResyncReplicationAction extends TransportWriteAction<Resyn
         if (node.getVersion().onOrAfter(Version.V_6_0_0_alpha1)) {
             super.sendReplicaRequest(replicaRequest, node, listener);
         } else {
-            listener.onResponse(new ReplicaResponse(SequenceNumbersService.UNASSIGNED_SEQ_NO));
+            listener.onResponse(new ReplicaResponse(SequenceNumbersService.PRE_60_NODE_LOCAL_CHECKPOINT));
         }
     }
 
diff --git a/core/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java b/core/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java
index e885225cb80..ae245fe252c 100644
--- a/core/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java
+++ b/core/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java
@@ -1023,6 +1023,12 @@ public abstract class TransportReplicationAction<
         }
 
         public ReplicaResponse(long localCheckpoint) {
+            /*
+             * A replica should always know its own local checkpoint so this should always be a valid sequence number or the pre-6.0 local
+             * checkpoint value when simulating responses to replication actions that pre-6.0 nodes are not aware of (e.g., the global
+             * checkpoint background sync, and the primary/replica resync).
+             */
+            assert localCheckpoint != SequenceNumbersService.UNASSIGNED_SEQ_NO;
             this.localCheckpoint = localCheckpoint;
         }
 
diff --git a/core/src/main/java/org/elasticsearch/index/seqno/GlobalCheckpointSyncAction.java b/core/src/main/java/org/elasticsearch/index/seqno/GlobalCheckpointSyncAction.java
index 897dc9beb32..c7059d05919 100644
--- a/core/src/main/java/org/elasticsearch/index/seqno/GlobalCheckpointSyncAction.java
+++ b/core/src/main/java/org/elasticsearch/index/seqno/GlobalCheckpointSyncAction.java
@@ -89,7 +89,7 @@ public class GlobalCheckpointSyncAction extends TransportReplicationAction<
         if (node.getVersion().onOrAfter(Version.V_6_0_0_alpha1)) {
             super.sendReplicaRequest(replicaRequest, node, listener);
         } else {
-            listener.onResponse(new ReplicaResponse(SequenceNumbersService.UNASSIGNED_SEQ_NO));
+            listener.onResponse(new ReplicaResponse(SequenceNumbersService.PRE_60_NODE_LOCAL_CHECKPOINT));
         }
     }
 
diff --git a/core/src/main/java/org/elasticsearch/index/seqno/GlobalCheckpointTracker.java b/core/src/main/java/org/elasticsearch/index/seqno/GlobalCheckpointTracker.java
index c112e82998b..447815cf9af 100644
--- a/core/src/main/java/org/elasticsearch/index/seqno/GlobalCheckpointTracker.java
+++ b/core/src/main/java/org/elasticsearch/index/seqno/GlobalCheckpointTracker.java
@@ -450,6 +450,9 @@ public class GlobalCheckpointTracker extends AbstractIndexShardComponent {
         assert lcps.localCheckpoint != SequenceNumbersService.PRE_60_NODE_LOCAL_CHECKPOINT ||
             localCheckpoint == SequenceNumbersService.PRE_60_NODE_LOCAL_CHECKPOINT :
             "pre-6.0 shard copy " + allocationId + " unexpected to send valid local checkpoint " + localCheckpoint;
+        // a local checkpoint for a shard copy should be a valid sequence number or the pre-6.0 sequence number indicator
+        assert localCheckpoint != SequenceNumbersService.UNASSIGNED_SEQ_NO :
+                "invalid local checkpoint for shard copy [" + allocationId + "]";
         if (localCheckpoint > lcps.localCheckpoint) {
             logger.trace("updated local checkpoint of [{}] from [{}] to [{}]", allocationId, lcps.localCheckpoint, localCheckpoint);
             lcps.localCheckpoint = localCheckpoint;
diff --git a/core/src/test/java/org/elasticsearch/index/seqno/GlobalCheckpointTrackerTests.java b/core/src/test/java/org/elasticsearch/index/seqno/GlobalCheckpointTrackerTests.java
index f06ffc0d45f..2f7d2dd15ce 100644
--- a/core/src/test/java/org/elasticsearch/index/seqno/GlobalCheckpointTrackerTests.java
+++ b/core/src/test/java/org/elasticsearch/index/seqno/GlobalCheckpointTrackerTests.java
@@ -737,7 +737,7 @@ public class GlobalCheckpointTrackerTests extends ESTestCase {
     private static void randomLocalCheckpointUpdate(GlobalCheckpointTracker gcp) {
         String allocationId = randomFrom(gcp.localCheckpoints.keySet());
         long currentLocalCheckpoint = gcp.localCheckpoints.get(allocationId).getLocalCheckpoint();
-        gcp.updateLocalCheckpoint(allocationId, currentLocalCheckpoint + randomInt(5));
+        gcp.updateLocalCheckpoint(allocationId, Math.max(SequenceNumbersService.NO_OPS_PERFORMED, currentLocalCheckpoint + randomInt(5)));
     }
 
     private static void randomMarkInSync(GlobalCheckpointTracker gcp) {