[Segment Replication] Ingestion round removes previously ingested documents

[dreamer-89]

Describe the bug
While working on #5946, identified scenario where documents ingested in current round removes all documents ingested previously. This might be potential data loss issue, so filing the bug.

To Reproduce
This issue appears pretty reliably with below integration test.

    public void testMultipleIngestions() throws Exception {
        final String primary = internalCluster().startNode();
        createIndex(INDEX_NAME);
        final String replica = internalCluster().startNode();
        ensureGreen(INDEX_NAME);

        // First round of ingestion
        final int initialDocCount = 500;
        ingestDocs(initialDocCount);
        refresh(INDEX_NAME);
        waitForReplicaUpdate();
        assertDocCounts(initialDocCount, replica, primary); // Test assertion passes here i.e. documents are searchable at this point

        // Second round of ingestion
        ingestDocs(2 * initialDocCount);
        ensureGreen(INDEX_NAME);
        flushAndRefresh(INDEX_NAME);
        waitForReplicaUpdate();
        assertDocCounts(3 * initialDocCount, replica, primary); // --> Test fails here with actual document count = 2 * initialdocCount
    }

Expected behavior
The total document count should match with documents ingested.

Host/Environment (please complete the following information):

• OS: iOS
• Version: main

[mch2]

@dreamer-89. I'm not sure what the implementation of ingestDocs is doing in this test, is it tracking the id used between rounds guaranteeing they are unique? This test is ingesting 500 docs, waiting for replica to update then ingesting 1000 docs. Based on the result it looks to me that it is resetting the id each time, so i'd expect only 1k docs, with 500 goign through an update. Perhaps you can check latest op seqno at the end?

[mch2]

I think its from SegmentReplicationSnapshotIT? In which case every invocation will create a new bg indexer starting from 0 doc id.

    public void ingestData(int docCount, String indexName) throws Exception {
        try (
            BackgroundIndexer indexer = new BackgroundIndexer(
                indexName,
                "_doc",
                client(),
                -1,
                RandomizedTest.scaledRandomIntBetween(2, 5),
                false,
                random()
            )
        ) {
            indexer.start(docCount);
            waitForDocs(docCount, indexer);
            refresh(indexName);
        }
    }