<commit_before>/*
 * Licensed to Elasticsearch under one or more contributor
 * license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright
 * ownership. Elasticsearch licenses this file to you under
 * the Apache License, Version 2.0 (the "License"); you may
 * not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.elasticsearch.versioning;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.elasticsearch.ExceptionsHelper;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.cluster.coordination.LinearizabilityChecker;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.SuppressForbidden;
import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.io.stream.BytesStreamOutput;
import org.elasticsearch.common.io.stream.InputStreamStreamInput;
import org.elasticsearch.common.io.stream.NamedWriteable;
import org.elasticsearch.common.io.stream.NamedWriteableAwareStreamInput;
import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.discovery.AbstractDisruptionTestCase;
import org.elasticsearch.index.engine.VersionConflictEngineException;
import org.elasticsearch.test.ESIntegTestCase;
import org.elasticsearch.test.disruption.ServiceDisruptionScheme;
import org.elasticsearch.test.junit.annotations.TestLogging;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Base64;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThan;


/**
 * This test stress tests CAS updates using sequence number based versioning (ifPrimaryTerm/ifSeqNo).
 *
 * <p>The following is a summary of the expected CAS write behaviour of the system:</p>
 *
 * <ul>
 *     <li>acknowledged CAS writes are guaranteed to have taken place between invocation and response and cannot be lost. It is
 *     guaranteed that the previous value had the specified primaryTerm and seqNo</li>
 *     <li>CAS writes resulting in a VersionConflictEngineException might or might not have taken place or may take place in the future
 *     provided the primaryTerm and seqNo still matches. The reason we cannot assume it will not take place after receiving the failure
 *     is that a request can fork into two because of retries on disconnect, and now race against itself. The retry might complete (and do a
 *     dirty or stale read) before the forked off request gets to execute, and that one might still subsequently succeed.
 *
 *     Such writes are not necessarily fully replicated and can be lost. There is no
 *     guarantee that the previous value did not have the specified primaryTerm and seqNo</li>
 *     <li>CAS writes with other exceptions might or might not have taken place. If they have taken place, then after invocation but not
 *     necessarily before response. Such writes are not necessarily fully replicated and can be lost.
 *     </li>
 * </ul>
 *
 * A deeper technical explanation of the behaviour is given here:
 *
 * <ul>
 *     <li>A CAS can fail on its own write in at least two ways. In both cases, the write might have taken place even though we get a
 *     version conflict response. Even though we might observe the write (by reading (not done in this test) or another CAS write), the
 *     write could be lost since it is not fully replicated. Details:
 *     <ul>
 *         <li>A write is successfully stored on primary and one replica (r1). Replication to second replica fails, primary is demoted
 *         and r1 is promoted to primary. The request is repeated on r1, but this time the request fails due to its own write.</li>
 *         <li>A coordinator sends write to primary, which stores write successfully (and replicates it). Connection is lost before
 *         response is sent back. Once connection is back, coordinator will retry against either same or new primary, but this time the
 *         request will fail due to its own write.
 *         </li>
 *     </ul>
 *     </li>
 *     <li>A CAS can fail on stale reads. A CAS failure is only checked on the supposedly primary node. However, the primary might not be
 *     the newest primary (could be isolated or just not have been told yet). So a CAS check is suspect to stale reads (like any
 *     read) and can thus fail due to reading stale data. Notice that a CAS success is fully replicated and thus guaranteed to not
 *     suffer from stale (or dirty) reads.
 *     </li>
 *     <li>A CAS can fail on a dirty read, i.e., a non-replicated write that ends up being discarded.</li>
 *     <li>For any other failure, we do not know if the write will succeed after the failure. However, we do know that if we
 *     subsequently get back a CAS success with seqNo s, any previous failures with ifSeqNo &lt; s will not be able to succeed (but could
 *     produce dirty writes on a stale primary).
 *     </li>
 *     <li>A CAS failure or any other failure can eventually succeed after receiving the failure response due to reroute and retries,
 *     see above.</li>
 *     <li>A CAS failure throws a VersionConflictEngineException which does not directly contain the current seqno/primary-term to use for
 *     the next request. It is contained in the message (and we parse it out in the test), but notice that the numbers given here could be
 *     stale or dirty, i.e., come from a stale primary or belong to a write that ends up being discarded.</li>
 * </ul>
 */
@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, minNumDataNodes = 4, maxNumDataNodes = 6)
@TestLogging("_root:DEBUG,org.elasticsearch.action.bulk:TRACE,org.elasticsearch.action.get:TRACE," +
    "org.elasticsearch.discovery:TRACE,org.elasticsearch.action.support.replication:TRACE," +
    "org.elasticsearch.cluster.service:TRACE,org.elasticsearch.indices.recovery:TRACE," +
    "org.elasticsearch.indices.cluster:TRACE,org.elasticsearch.index.shard:TRACE")
public class ConcurrentSeqNoVersioningIT extends AbstractDisruptionTestCase {

    private static final Pattern EXTRACT_VERSION = Pattern.compile("current document has seqNo \\[(\\d+)\\] and primary term \\[(\\d+)\\]");

    // Test info: disrupt network for up to 8s in a number of rounds and check that we only get true positive CAS results when running
    // multiple threads doing CAS updates.
    // Wait up to 1 minute (+10s in thread to ensure it does not time out) for threads to complete previous round before initiating next
    // round.
    public void testSeqNoCASLinearizability() {
        final int disruptTimeSeconds = scaledRandomIntBetween(1, 8);

        assertAcked(prepareCreate("test")
            .setSettings(Settings.builder()
                .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1 + randomInt(2))
                .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, randomInt(3))
            ));

        ensureGreen();

        int numberOfKeys = randomIntBetween(1, 10);

        logger.info("--> Indexing initial doc for {} keys", numberOfKeys);
        List<Partition> partitions =
            IntStream.range(0, numberOfKeys)
                .mapToObj(i -> client().prepareIndex("test", "type", "ID:" + i).setSource("value", -1).get())
                .map(response ->
                    new Partition(response.getId(), new Version(response.getPrimaryTerm(), response.getSeqNo())))
                .collect(Collectors.toList());

        int threadCount = randomIntBetween(3, 20);
        CyclicBarrier roundBarrier = new CyclicBarrier(threadCount + 1); // +1 for main thread.
<commit_msg>Use correct global checkpoint sync interval (#42642)<commit_after>/*
 * Licensed to Elasticsearch under one or more contributor
 * license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright
 * ownership. Elasticsearch licenses this file to you under
 * the Apache License, Version 2.0 (the "License"); you may
 * not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.elasticsearch.versioning;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.elasticsearch.ExceptionsHelper;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.cluster.coordination.LinearizabilityChecker;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.SuppressForbidden;
import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.io.stream.BytesStreamOutput;
import org.elasticsearch.common.io.stream.InputStreamStreamInput;
import org.elasticsearch.common.io.stream.NamedWriteable;
import org.elasticsearch.common.io.stream.NamedWriteableAwareStreamInput;
import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.discovery.AbstractDisruptionTestCase;
import org.elasticsearch.index.engine.VersionConflictEngineException;
import org.elasticsearch.test.ESIntegTestCase;
import org.elasticsearch.test.disruption.ServiceDisruptionScheme;
import org.elasticsearch.test.junit.annotations.TestLogging;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Base64;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThan;


/**
 * This test stress tests CAS updates using sequence number based versioning (ifPrimaryTerm/ifSeqNo).
 *
 * <p>The following is a summary of the expected CAS write behaviour of the system:</p>
 *
 * <ul>
 *     <li>acknowledged CAS writes are guaranteed to have taken place between invocation and response and cannot be lost. It is
 *     guaranteed that the previous value had the specified primaryTerm and seqNo</li>
 *     <li>CAS writes resulting in a VersionConflictEngineException might or might not have taken place or may take place in the future
 *     provided the primaryTerm and seqNo still matches. The reason we cannot assume it will not take place after receiving the failure
 *     is that a request can fork into two because of retries on disconnect, and now race against itself. The retry might complete (and do a
 *     dirty or stale read) before the forked off request gets to execute, and that one might still subsequently succeed.
 *
 *     Such writes are not necessarily fully replicated and can be lost. There is no
 *     guarantee that the previous value did not have the specified primaryTerm and seqNo</li>
 *     <li>CAS writes with other exceptions might or might not have taken place. If they have taken place, then after invocation but not
 *     necessarily before response. Such writes are not necessarily fully replicated and can be lost.
 *     </li>
 * </ul>
 *
 * A deeper technical explanation of the behaviour is given here:
 *
 * <ul>
 *     <li>A CAS can fail on its own write in at least two ways. In both cases, the write might have taken place even though we get a
 *     version conflict response. Even though we might observe the write (by reading (not done in this test) or another CAS write), the
 *     write could be lost since it is not fully replicated. Details:
 *     <ul>
 *         <li>A write is successfully stored on primary and one replica (r1). Replication to second replica fails, primary is demoted
 *         and r1 is promoted to primary. The request is repeated on r1, but this time the request fails due to its own write.</li>
 *         <li>A coordinator sends write to primary, which stores write successfully (and replicates it). Connection is lost before
 *         response is sent back. Once connection is back, coordinator will retry against either same or new primary, but this time the
 *         request will fail due to its own write.
 *         </li>
 *     </ul>
 *     </li>
 *     <li>A CAS can fail on stale reads. A CAS failure is only checked on the supposedly primary node. However, the primary might not be
 *     the newest primary (could be isolated or just not have been told yet). So a CAS check is suspect to stale reads (like any
 *     read) and can thus fail due to reading stale data. Notice that a CAS success is fully replicated and thus guaranteed to not
 *     suffer from stale (or dirty) reads.
 *     </li>
 *     <li>A CAS can fail on a dirty read, i.e., a non-replicated write that ends up being discarded.</li>
 *     <li>For any other failure, we do not know if the write will succeed after the failure. However, we do know that if we
 *     subsequently get back a CAS success with seqNo s, any previous failures with ifSeqNo &lt; s will not be able to succeed (but could
 *     produce dirty writes on a stale primary).
 *     </li>
 *     <li>A CAS failure or any other failure can eventually succeed after receiving the failure response due to reroute and retries,
 *     see above.</li>
 *     <li>A CAS failure throws a VersionConflictEngineException which does not directly contain the current seqno/primary-term to use for
 *     the next request. It is contained in the message (and we parse it out in the test), but notice that the numbers given here could be
 *     stale or dirty, i.e., come from a stale primary or belong to a write that ends up being discarded.</li>
 * </ul>
 */
@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, minNumDataNodes = 4, maxNumDataNodes = 6)
@TestLogging("_root:DEBUG,org.elasticsearch.action.bulk:TRACE,org.elasticsearch.action.get:TRACE," +
    "org.elasticsearch.discovery:TRACE,org.elasticsearch.action.support.replication:TRACE," +
    "org.elasticsearch.cluster.service:TRACE,org.elasticsearch.indices.recovery:TRACE," +
    "org.elasticsearch.indices.cluster:TRACE,org.elasticsearch.index.shard:TRACE")
public class ConcurrentSeqNoVersioningIT extends AbstractDisruptionTestCase {

    private static final Pattern EXTRACT_VERSION = Pattern.compile("current document has seqNo \\[(\\d+)\\] and primary term \\[(\\d+)\\]");

    // Test info: disrupt network for up to 8s in a number of rounds and check that we only get true positive CAS results when running
    // multiple threads doing CAS updates.
    // Wait up to 1 minute (+10s in thread to ensure it does not time out) for threads to complete previous round before initiating next
    // round.
    public void testSeqNoCASLinearizability() {
        final int disruptTimeSeconds = scaledRandomIntBetween(1, 8);

        assertAcked(prepareCreate("test")
            .setSettings(Settings.builder()
                .put(indexSettings())
                .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1 + randomInt(2))
                .put(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, randomInt(3))
            ));

        ensureGreen();

        int numberOfKeys = randomIntBetween(1, 10);

        logger.info("--> Indexing initial doc for {} keys", numberOfKeys);
        List<Partition> partitions =
            IntStream.range(0, numberOfKeys)
                .mapToObj(i -> client().prepareIndex("test", "type", "ID:" + i).setSource("value", -1).get())
                .map(response ->
                    new Partition(response.getId(), new Version(response.getPrimaryTerm(), response.getSeqNo())))
                .collect(Collectors.toList());

        int threadCount = randomIntBetween(3, 20);
        CyclicBarrier roundBarrier = new CyclicBarrier(threadCount + 1); // +1 for main thread.

{'fraction_non_alphanumeric': 0.07891368467916968, 'fraction_numerical': 0.003728643775390951, 'mean_word_length': 3.8580697485806974, 'pattern_counts': {'":': 0, '<': 65, '<?xml version=': 0, '>': 71, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7663873', 'n_tokens_mistral': 4702, 'n_tokens_neox': 4610, 'n_words': 1905}