Prevent CCR recovery from missing documents

[Tim-Brooks]

Currently the snapshot/restore process manually sets the global
checkpoint to the max sequence number from the restored segements. This
does not work for Ccr as this will lead to documents that would be
recovered in the normal followering operation from being recovered.

This commit fixes this issue by setting the initial global checkpoint to
the existing local checkpoint.

[elasticmachine]

Pinging @elastic/es-distributed

[Tim-Brooks]

This is not ready for production. I pushed it up here so we can discuss.

This commit:

1. Adds a test that will occasionally fail due to the local checkpoint != the max sequence number. Currently the test is set to run 5 times to guarantee consistent failures. However, that will be removed eventually.

2. This commit comments out store.bootstrapNewHistory(). This method manually sets the the local checkpoint == the max sequence number. We need to discuss how to deal with this as I think the bootstrap new history method probably offers some behavior we want to retain (even if the sequence number part does not work for CCR).

3. This commit attempts to advance the global checkpoint to the local check point after putting no-ops (for the normal non-ccr restore process). However, that cannot happen here as the replicationTracker.isPrimaryMode() is not yet in primary mode and that triggers an assertion. I'm not sure if we need to advance the global checkpoint in StoreRecovery#restore as I see that when the replication tracker goes to primary mode, it will advance the global checkpoint there. But this is something we should discuss.

    /**
     * Initializes the global checkpoint tracker in primary mode (see {@link #primaryMode}. Called on primary activation or promotion.
     */
    public synchronized void activatePrimaryMode(final long localCheckpoint) {
        assert invariant();
        assert primaryMode == false;
        assert checkpoints.get(shardAllocationId) != null && checkpoints.get(shardAllocationId).inSync &&
            checkpoints.get(shardAllocationId).localCheckpoint == SequenceNumbers.UNASSIGNED_SEQ_NO :
            "expected " + shardAllocationId + " to have initialized entry in " + checkpoints + " when activating primary";
        assert localCheckpoint >= SequenceNumbers.NO_OPS_PERFORMED;
        primaryMode = true;
        updateLocalCheckpoint(shardAllocationId, checkpoints.get(shardAllocationId), localCheckpoint);
        updateGlobalCheckpointOnPrimary();
        assert invariant();
    }

[ywelsch]

@tbrooks8 I've fixed things up as far as I think how we should handle this. I've also added a unit test. The integration test testFollowIndexWithConcurrentMappingChanges looks to fail sometimes, unrelated to the changes I've made here. It has some problems with concurrent mapping updates (where it tries to both update a field as text as well as a long).

While this fixes the problem here, it exposes another issue I think, namely that the primary will start off (i.e. be marked as started) with a history that contains gaps, i.e. local checkpoint != max sequence number. This can turn out to be problematic for replicas, because peer recovery only completes if all gaps are filled on the primary (see call to cancellableThreads.execute(() -> shard.waitForOpsToComplete(endingSeqNo)); in RecoverySourceHandler). This means that if someone does a PUT FOLLOW with wait_for_active_shards > 1, and we restore an IndexCommit with gaps, then the wait condition will time out, as we only start the following (which will complete the gaps) once the wait condition has passed. I think we should have initiateFollowing to first resume the following, and then execute the wait condition. This problem should be verifiable by extending the existing test with a wait_for_active_shards > 1 condition.

[ywelsch]

LGTM

[Tim-Brooks]

@ywelsch your changes look good to me. There is a test failing (reproducibly). Looks like we assert that the sequence numbers match between Lucene and the translog. Did you mean for this to be the assertion for testRestoreShard:

closeShard(target, false);
closeShards(source);

instead of:

closeShard(source, false);
closeShards(target);

I assume we should not be asserting the ops are the same for the target since there will be some no-ops? I can update the PR, I just wanted to check if that is what you intended.

[ywelsch]

No, I only wanted the source index to be leniently closed (because we force-fully inject a gap). The problem was that the test was not ensuring that the index is restored with soft-deletes enabled when it is snapshotted with soft-deletes enabled. I've fixed this now

[ywelsch]

@elasticmachine run elasticsearch-ci/default-distro

[ywelsch]

@elasticmachine run elasticsearch-ci/2