Resync should not send operations without sequence number

server/src/main/java/org/elasticsearch/index/shard/PrimaryReplicaSyncer.java

@@ -246,11 +246,12 @@ protected void doRun() throws Exception {
             Translog.Operation operation;
             while ((operation = snapshot.next()) != null) {
                 final long seqNo = operation.seqNo();
-                if (startingSeqNo >= 0 &&
-                    (seqNo == SequenceNumbers.UNASSIGNED_SEQ_NO || seqNo < startingSeqNo)) {
+                // do not send operations if the global checkpoint is still unassigned.
+                if (seqNo == SequenceNumbers.UNASSIGNED_SEQ_NO || startingSeqNo < 0 || seqNo < startingSeqNo) {
                     totalSkippedOps.incrementAndGet();
                     continue;
                 }
+                assert operation.seqNo() >= 0 : "sending operation with unassigned sequence number [" + operation + "]";
                 operations.add(operation);
                 size += operation.estimateSize();
                 totalSentOps.incrementAndGet();
@@ -260,7 +261,6 @@ protected void doRun() throws Exception {
                     break;
                 }
             }
-
             final long trimmedAboveSeqNo = firstMessage.get() ? maxSeqNo : SequenceNumbers.UNASSIGNED_SEQ_NO;
             // have to send sync request even in case of there are no operations to sync - have to sync trimmedAboveSeqNo at least
             if (!operations.isEmpty() || trimmedAboveSeqNo != SequenceNumbers.UNASSIGNED_SEQ_NO) {

server/src/test/java/org/elasticsearch/index/shard/PrimaryReplicaSyncerTests.java

@@ -42,6 +42,8 @@
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.index.mapper.SourceToParse;
 import org.elasticsearch.index.seqno.SequenceNumbers;
+import org.elasticsearch.index.translog.TestTranslog;
+import org.elasticsearch.index.translog.Translog;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskManager;
 
@@ -54,9 +56,16 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.emptyArray;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.core.IsInstanceOf.instanceOf;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
 
 public class PrimaryReplicaSyncerTests extends IndexShardTestCase {
 
@@ -186,6 +195,37 @@ public void onResponse(PrimaryReplicaSyncer.ResyncTask result) {
         }
     }
 
+    public void testDoNotSendOperationsIfGlobalCheckpointUnassigned() throws Exception {
+        IndexShard shard = spy(newStartedShard(true));
+        when(shard.getGlobalCheckpoint()).thenReturn(SequenceNumbers.UNASSIGNED_SEQ_NO);
+        int numOps = between(0, 20);
+        List<Translog.Operation> operations = new ArrayList<>();
+        for (int i = 0; i < numOps; i++) {
+            operations.add(new Translog.Index(
+                "_doc", Integer.toString(i), randomBoolean() ? SequenceNumbers.UNASSIGNED_SEQ_NO : i, primaryTerm, new byte[]{1}));
+        }
+        doReturn(TestTranslog.newSnapshotFromOperations(operations)).when(shard).getHistoryOperations(anyString(), anyLong());
+        TaskManager taskManager = new TaskManager(Settings.EMPTY, threadPool, Collections.emptySet());
+        List<ResyncReplicationRequest> resyncRequests = new ArrayList<>();
+        PrimaryReplicaSyncer.SyncAction syncAction = (request, parentTask, allocationId, primaryTerm, listener) -> {
+            assertThat("primary-replica resync with unassigned global checkpoint should not send translog operations",
+                request.getOperations(), emptyArray());
+            resyncRequests.add(request);
+            listener.onResponse(new ResyncReplicationResponse());
+        };
+        PrimaryReplicaSyncer syncer = new PrimaryReplicaSyncer(taskManager, syncAction);
+        syncer.setChunkSize(new ByteSizeValue(randomIntBetween(1, 10)));
+        PlainActionFuture<PrimaryReplicaSyncer.ResyncTask> fut = new PlainActionFuture<>();
+        syncer.resync(shard, fut);
+        PrimaryReplicaSyncer.ResyncTask resyncTask = fut.actionGet();
+        assertThat(resyncRequests, hasSize(1));
+        assertThat(resyncRequests.get(0).getOperations(), emptyArray());
+        assertThat(resyncTask.getSkippedOperations(), equalTo(numOps));
+        assertThat(resyncTask.getResyncedOperations(), equalTo(0));
+        assertThat(resyncTask.getTotalOperations(), equalTo(numOps));
+        closeShards(shard);
+    }
+
     public void testStatusSerialization() throws IOException {
         PrimaryReplicaSyncer.ResyncTask.Status status = new PrimaryReplicaSyncer.ResyncTask.Status(randomAlphaOfLength(10),
             randomIntBetween(0, 1000), randomIntBetween(0, 1000), randomIntBetween(0, 1000));

server/src/test/java/org/elasticsearch/index/translog/TestTranslog.java

@@ -37,6 +37,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Comparator;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Random;
 import java.util.Set;
@@ -142,4 +143,28 @@ public static List<Translog.Operation> drainSnapshot(Translog.Snapshot snapshot,
         }
         return ops;
     }
+
+    public static Translog.Snapshot newSnapshotFromOperations(List<Translog.Operation> operations) {
+        final Iterator<Translog.Operation> iterator = operations.iterator();
+        return new Translog.Snapshot() {
+            @Override
+            public int totalOperations() {
+                return operations.size();
+            }
+
+            @Override
+            public Translog.Operation next() {
+                if (iterator.hasNext()) {
+                    return iterator.next();
+                } else {
+                    return null;
+                }
+            }
+
+            @Override
+            public void close() {
+
+            }
+        };
+    }
 }