/*
 * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
 * or more contributor license agreements. Licensed under the Elastic License;
 * you may not use this file except in compliance with the Elastic License.
 */
package org.elasticsearch.xpack.ccr.action;

import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.cluster.routing.ShardRoutingState;
import org.elasticsearch.cluster.routing.TestShardRouting;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.IndexService;
import org.elasticsearch.index.shard.IndexShard;
import org.elasticsearch.index.shard.IndexShardNotStartedException;
import org.elasticsearch.index.translog.Translog;
import org.elasticsearch.test.ESSingleNodeTestCase;
import org.mockito.Mockito;

import java.util.List;

import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;

public class ShardChangesActionTests extends ESSingleNodeTestCase {

    public void testGetOperationsBetween() throws Exception {
        final Settings settings = Settings.builder()
                .put("index.number_of_shards", 1)
                .put("index.number_of_replicas", 0)
                .put("index.translog.generation_threshold_size", new ByteSizeValue(randomIntBetween(8, 64), ByteSizeUnit.KB))
                .build();
        final IndexService indexService = createIndex("index", settings);

        final int numWrites = randomIntBetween(2, 8192);
        for (int i = 0; i < numWrites; i++) {
            client().prepareIndex("index", "doc", Integer.toString(i)).setSource("{}", XContentType.JSON).get();
        }

        // A number of times, get operations within a range that exists:
        int iters = randomIntBetween(8, 32);
        IndexShard indexShard = indexService.getShard(0);
        for (int iter = 0; iter < iters; iter++) {
            int min = randomIntBetween(0, numWrites - 1);
            int max = randomIntBetween(min, numWrites - 1);

            int index = 0;
            List<Translog.Operation> operations = ShardChangesAction.getOperationsBetween(indexShard, min, max);
            for (long expectedSeqNo = min; expectedSeqNo < max; expectedSeqNo++) {
                Translog.Operation operation = operations.get(index++);
                assertThat(operation.seqNo(), equalTo(expectedSeqNo));
            }
        }

        // get operations for a range no operations exists:
        Exception e = expectThrows(IllegalStateException.class,
                () -> ShardChangesAction.getOperationsBetween(indexShard, numWrites, numWrites + 1));
        assertThat(e.getMessage(), containsString("Not all operations between min_seq_no [" + numWrites + "] and max_seq_no [" +
                (numWrites + 1) +"] found, tracker checkpoint ["));

        // get operations for a range some operations do not exist:
        e = expectThrows(IllegalStateException.class,
                () -> ShardChangesAction.getOperationsBetween(indexShard, numWrites  - 10, numWrites + 10));
        assertThat(e.getMessage(), containsString("Not all operations between min_seq_no [" + (numWrites - 10) + "] and max_seq_no [" +
                (numWrites + 10) +"] found, tracker checkpoint ["));
    }

    public void testGetOperationsBetweenWhenShardNotStarted() throws Exception {
        IndexShard indexShard = Mockito.mock(IndexShard.class);

        ShardRouting shardRouting = TestShardRouting.newShardRouting("index", 0, "_node_id", true, ShardRoutingState.INITIALIZING);
        Mockito.when(indexShard.routingEntry()).thenReturn(shardRouting);
        expectThrows(IndexShardNotStartedException.class, () -> ShardChangesAction.getOperationsBetween(indexShard, 0, 1));
    }

}

@@ -978,16 +978,110 @@
 il.List;
+%0Aimport java.util.Set;%0Aimport java.util.stream.Collectors;%0Aimport java.util.stream.LongStream;
 %0A%0Aimport
@@ -2274,34 +2274,13 @@
     
+f
 in
-t index = 0;%0A           
+al
  Lis
@@ -2393,155 +2393,491 @@
     
-for (long expectedSeqNo = min; expectedSeqNo %3C max; expectedSeqNo++) %7B%0A                Translog.Operation operation = operations.get(index++);%0A    
+/*%0A             * We are not guaranteed that operations are returned to us in order they are in the translog (if our read crosses multiple%0A             * generations) so the best we can assert is that we see the expected operations.%0A             */%0A            final Set%3CLong%3E seenSeqNos = operations.stream().map(Translog.Operation::seqNo).collect(Collectors.toSet());%0A            final Set%3CLong%3E expectedSeqNos = LongStream.range(min, max + 1).boxed().collect(Collectors.toSet());%0A
     
@@ -2899,25 +2899,18 @@
 hat(
-operation.s
+seenS
 eqNo
-()
+s
 , eq
@@ -2932,26 +2932,13 @@
 eqNo
+s
 ));%0A
-            %7D%0A
     

Fix get operations between test