Merge remote-tracking branch 'origin/main' into dev/PublishCheckpointAction_use_never_give_up_retry_strategy

Author: guojialiang92

CHANGELOG.md

@@ -40,6 +40,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add TermsQuery support to Search GRPC endpoint ([#17888](https://github.com/opensearch-project/OpenSearch/pull/17888))
 - Support sub agg in filter rewrite optimization ([#17447](https://github.com/opensearch-project/OpenSearch/pull/17447)
 - Disable scoring of keyword term search by default, fallback logic with new use_similarity:true parameter ([#17889](https://github.com/opensearch-project/OpenSearch/pull/17889))
+- Add versioning support in pull-based ingestion ([#17918](https://github.com/opensearch-project/OpenSearch/pull/17918))
 
 ### Changed
 - Migrate BC libs to their FIPS counterparts ([#14912](https://github.com/opensearch-project/OpenSearch/pull/14912))

plugins/ingestion-kafka/src/internalClusterTest/java/org/opensearch/plugin/kafka/KafkaIngestionBaseIT.java

@@ -108,6 +108,19 @@ protected void produceData(String id, String name, String age, long timestamp, S
         producer.send(new ProducerRecord<>(topicName, null, timestamp, "null", payload));
     }
 
+    protected void produceDataWithExternalVersion(String id, long version, String name, String age, long timestamp, String opType) {
+        String payload = String.format(
+            Locale.ROOT,
+            "{\"_id\":\"%s\", \"_version\":\"%d\", \"_op_type\":\"%s\",\"_source\":{\"name\":\"%s\", \"age\": %s}}",
+            id,
+            version,
+            opType,
+            name,
+            age
+        );
+        producer.send(new ProducerRecord<>(topicName, null, timestamp, "null", payload));
+    }
+
     protected void produceData(String payload) {
         producer.send(new ProducerRecord<>(topicName, null, defaultMessageTimestamp, "null", payload));
     }

plugins/ingestion-kafka/src/internalClusterTest/java/org/opensearch/plugin/kafka/RemoteStoreKafkaIT.java

@@ -19,7 +19,9 @@
 import org.opensearch.cluster.metadata.IndexMetadata;
 import org.opensearch.cluster.routing.allocation.command.AllocateReplicaAllocationCommand;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.index.query.BoolQueryBuilder;
 import org.opensearch.index.query.RangeQueryBuilder;
+import org.opensearch.index.query.TermQueryBuilder;
 import org.opensearch.test.InternalTestCluster;
 import org.opensearch.test.OpenSearchIntegTestCase;
 import org.opensearch.transport.client.Requests;
@@ -310,6 +312,122 @@ public void testPaginatedGetIngestionState() throws ExecutionException, Interrup
         }));
     }
 
+    public void testExternalVersioning() throws Exception {
+        // setup nodes and index
+        produceDataWithExternalVersion("1", 1, "name1", "25", defaultMessageTimestamp, "index");
+        produceDataWithExternalVersion("2", 1, "name2", "25", defaultMessageTimestamp, "index");
+        internalCluster().startClusterManagerOnlyNode();
+        final String nodeA = internalCluster().startDataOnlyNode();
+        final String nodeB = internalCluster().startDataOnlyNode();
+
+        createIndexWithDefaultSettings(1, 1);
+        ensureGreen(indexName);
+        waitForSearchableDocs(2, Arrays.asList(nodeA, nodeB));
+
+        // validate next version docs get indexed
+        produceDataWithExternalVersion("1", 2, "name1", "30", defaultMessageTimestamp, "index");
+        produceDataWithExternalVersion("2", 2, "name2", "30", defaultMessageTimestamp, "index");
+        waitForState(() -> {
+            BoolQueryBuilder query1 = new BoolQueryBuilder().must(new TermQueryBuilder("_id", 1));
+            SearchResponse response1 = client().prepareSearch(indexName).setQuery(query1).get();
+            assertThat(response1.getHits().getTotalHits().value(), is(1L));
+            BoolQueryBuilder query2 = new BoolQueryBuilder().must(new TermQueryBuilder("_id", 2));
+            SearchResponse response2 = client().prepareSearch(indexName).setQuery(query2).get();
+            assertThat(response2.getHits().getTotalHits().value(), is(1L));
+            return 30 == (Integer) response1.getHits().getHits()[0].getSourceAsMap().get("age")
+                && 30 == (Integer) response2.getHits().getHits()[0].getSourceAsMap().get("age");
+        });
+
+        // test out-of-order updates
+        produceDataWithExternalVersion("1", 1, "name1", "25", defaultMessageTimestamp, "index");
+        produceDataWithExternalVersion("2", 1, "name2", "25", defaultMessageTimestamp, "index");
+        produceDataWithExternalVersion("3", 1, "name3", "25", defaultMessageTimestamp, "index");
+        waitForSearchableDocs(3, Arrays.asList(nodeA, nodeB));
+
+        BoolQueryBuilder query1 = new BoolQueryBuilder().must(new TermQueryBuilder("_id", 1));
+        SearchResponse response1 = client().prepareSearch(indexName).setQuery(query1).get();
+        assertThat(response1.getHits().getTotalHits().value(), is(1L));
+        assertEquals(30, response1.getHits().getHits()[0].getSourceAsMap().get("age"));
+
+        BoolQueryBuilder query2 = new BoolQueryBuilder().must(new TermQueryBuilder("_id", 2));
+        SearchResponse response2 = client().prepareSearch(indexName).setQuery(query2).get();
+        assertThat(response2.getHits().getTotalHits().value(), is(1L));
+        assertEquals(30, response2.getHits().getHits()[0].getSourceAsMap().get("age"));
+
+        // test deletes with smaller version
+        produceDataWithExternalVersion("1", 1, "name1", "25", defaultMessageTimestamp, "delete");
+        produceDataWithExternalVersion("4", 1, "name4", "25", defaultMessageTimestamp, "index");
+        waitForSearchableDocs(4, Arrays.asList(nodeA, nodeB));
+        RangeQueryBuilder query = new RangeQueryBuilder("age").gte(23);
+        SearchResponse response = client().prepareSearch(indexName).setQuery(query).get();
+        assertThat(response.getHits().getTotalHits().value(), is(4L));
+
+        // test deletes with correct version
+        produceDataWithExternalVersion("1", 3, "name1", "30", defaultMessageTimestamp, "delete");
+        produceDataWithExternalVersion("2", 3, "name2", "30", defaultMessageTimestamp, "delete");
+        waitForState(() -> {
+            RangeQueryBuilder rangeQuery = new RangeQueryBuilder("age").gte(23);
+            SearchResponse rangeQueryResponse = client().prepareSearch(indexName).setQuery(rangeQuery).get();
+            assertThat(rangeQueryResponse.getHits().getTotalHits().value(), is(2L));
+            return true;
+        });
+    }
+
+    public void testExternalVersioningWithDisabledGCDeletes() throws Exception {
+        // setup nodes and index
+        internalCluster().startClusterManagerOnlyNode();
+        final String nodeA = internalCluster().startDataOnlyNode();
+        final String nodeB = internalCluster().startDataOnlyNode();
+
+        createIndex(
+            indexName,
+            Settings.builder()
+                .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
+                .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1)
+                .put("ingestion_source.type", "kafka")
+                .put("ingestion_source.pointer.init.reset", "earliest")
+                .put("ingestion_source.param.topic", topicName)
+                .put("ingestion_source.param.bootstrap_servers", kafka.getBootstrapServers())
+                .put("index.replication.type", "SEGMENT")
+                .put("index.gc_deletes", "0")
+                .build(),
+            "{\"properties\":{\"name\":{\"type\": \"text\"},\"age\":{\"type\": \"integer\"}}}}"
+        );
+
+        // insert documents
+        produceDataWithExternalVersion("1", 1, "name1", "25", defaultMessageTimestamp, "index");
+        produceDataWithExternalVersion("2", 1, "name2", "25", defaultMessageTimestamp, "index");
+        waitForState(() -> {
+            RangeQueryBuilder rangeQuery = new RangeQueryBuilder("age").gte(23);
+            SearchResponse rangeQueryResponse = client().prepareSearch(indexName).setQuery(rangeQuery).get();
+            assertThat(rangeQueryResponse.getHits().getTotalHits().value(), is(2L));
+            return true;
+        });
+
+        // delete documents 1 and 2
+        produceDataWithExternalVersion("1", 2, "name1", "25", defaultMessageTimestamp, "delete");
+        produceDataWithExternalVersion("2", 2, "name2", "25", defaultMessageTimestamp, "delete");
+        produceDataWithExternalVersion("3", 1, "name3", "25", defaultMessageTimestamp, "index");
+        waitForState(() -> {
+            BoolQueryBuilder query = new BoolQueryBuilder().must(new TermQueryBuilder("_id", 3));
+            SearchResponse response = client().prepareSearch(indexName).setQuery(query).get();
+            assertThat(response.getHits().getTotalHits().value(), is(1L));
+            return 25 == (Integer) response.getHits().getHits()[0].getSourceAsMap().get("age");
+        });
+        waitForSearchableDocs(1, Arrays.asList(nodeA, nodeB));
+
+        // validate index operation with lower version creates new document
+        produceDataWithExternalVersion("1", 1, "name1", "35", defaultMessageTimestamp, "index");
+        produceDataWithExternalVersion("4", 1, "name4", "35", defaultMessageTimestamp, "index");
+        waitForState(() -> {
+            RangeQueryBuilder rangeQuery = new RangeQueryBuilder("age").gte(34);
+            SearchResponse rangeQueryResponse = client().prepareSearch(indexName).setQuery(rangeQuery).get();
+            assertThat(rangeQueryResponse.getHits().getTotalHits().value(), is(2L));
+            return true;
+        });
+
+    }
+
     private void verifyRemoteStoreEnabled(String node) {
         GetSettingsResponse settingsResponse = client(node).admin().indices().prepareGetSettings(indexName).get();
         String remoteStoreEnabled = settingsResponse.getIndexToSettings().get(indexName).get("index.remote_store.enabled");

server/src/internalClusterTest/java/org/opensearch/action/admin/indices/scale/searchonly/ScaleIndexIT.java

@@ -18,7 +18,6 @@
 import org.opensearch.cluster.routing.IndexShardRoutingTable;
 import org.opensearch.cluster.routing.ShardRouting;
 import org.opensearch.common.settings.Settings;
-import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.core.rest.RestStatus;
 import org.opensearch.indices.replication.common.ReplicationType;
 import org.opensearch.remotestore.RemoteStoreBaseIntegTestCase;
@@ -41,11 +40,6 @@ public class ScaleIndexIT extends RemoteStoreBaseIntegTestCase {
 
     private static final String TEST_INDEX = "test_scale_index";
 
-    @Override
-    protected Settings featureFlagSettings() {
-        return Settings.builder().put(super.featureFlagSettings()).put(FeatureFlags.READER_WRITER_SPLIT_EXPERIMENTAL, Boolean.TRUE).build();
-    }
-
     public Settings indexSettings() {
         return Settings.builder().put(SETTING_REPLICATION_TYPE, ReplicationType.SEGMENT).build();
     }

server/src/internalClusterTest/java/org/opensearch/cluster/allocation/SearchReplicaAllocationIT.java

@@ -12,7 +12,6 @@
 import org.opensearch.cluster.routing.IndexShardRoutingTable;
 import org.opensearch.cluster.routing.ShardRouting;
 import org.opensearch.common.settings.Settings;
-import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.indices.replication.common.ReplicationType;
 import org.opensearch.remotestore.RemoteStoreBaseIntegTestCase;
 import org.opensearch.test.OpenSearchIntegTestCase;
@@ -25,11 +24,6 @@
 @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0)
 public class SearchReplicaAllocationIT extends RemoteStoreBaseIntegTestCase {
 
-    @Override
-    protected Settings featureFlagSettings() {
-        return Settings.builder().put(super.featureFlagSettings()).put(FeatureFlags.READER_WRITER_SPLIT_EXPERIMENTAL, Boolean.TRUE).build();
-    }
-
     public void testSearchReplicaAllocatedToDedicatedSearchNode() {
         internalCluster().startClusterManagerOnlyNode();
         String primaryNode = internalCluster().startDataOnlyNode();

server/src/internalClusterTest/java/org/opensearch/cluster/allocation/SearchReplicaAwarenessAllocationIT.java

@@ -19,7 +19,6 @@
 import org.opensearch.cluster.routing.allocation.AwarenessReplicaBalance;
 import org.opensearch.cluster.routing.allocation.decider.AwarenessAllocationDecider;
 import org.opensearch.common.settings.Settings;
-import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.indices.replication.common.ReplicationType;
 import org.opensearch.remotestore.RemoteStoreBaseIntegTestCase;
 import org.opensearch.test.OpenSearchIntegTestCase;
@@ -40,11 +39,6 @@ public class SearchReplicaAwarenessAllocationIT extends RemoteStoreBaseIntegTest
 
     private final Logger logger = LogManager.getLogger(SearchReplicaAwarenessAllocationIT.class);
 
-    @Override
-    protected Settings featureFlagSettings() {
-        return Settings.builder().put(super.featureFlagSettings()).put(FeatureFlags.READER_WRITER_SPLIT_EXPERIMENTAL, Boolean.TRUE).build();
-    }
-
     public void testAllocationAwarenessZones() {
         Settings commonSettings = Settings.builder()
             .put(AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP_SETTING.getKey() + "zone.values", "a,b")

server/src/internalClusterTest/java/org/opensearch/cluster/metadata/AutoExpandSearchReplicasIT.java

@@ -11,7 +11,6 @@
 import org.opensearch.cluster.routing.UnassignedInfo;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
-import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.indices.replication.common.ReplicationType;
 import org.opensearch.remotestore.RemoteStoreBaseIntegTestCase;
 import org.opensearch.test.InternalTestCluster;
@@ -22,11 +21,6 @@
 @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0)
 public class AutoExpandSearchReplicasIT extends RemoteStoreBaseIntegTestCase {
 
-    @Override
-    protected Settings featureFlagSettings() {
-        return Settings.builder().put(super.featureFlagSettings()).put(FeatureFlags.READER_WRITER_SPLIT_EXPERIMENTAL, Boolean.TRUE).build();
-    }
-
     public void testAutoExpandSearchReplica() throws Exception {
         String indexName = "test";
         internalCluster().startClusterManagerOnlyNode();

server/src/internalClusterTest/java/org/opensearch/cluster/metadata/MetadataIndexStateServiceIT.java

@@ -11,7 +11,6 @@
 import org.opensearch.action.search.SearchResponse;
 import org.opensearch.action.support.WriteRequest;
 import org.opensearch.common.settings.Settings;
-import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.remotestore.RemoteStoreBaseIntegTestCase;
 import org.opensearch.test.OpenSearchIntegTestCase;
 
@@ -28,11 +27,6 @@ public class MetadataIndexStateServiceIT extends RemoteStoreBaseIntegTestCase {
 
     private static final String TEST_INDEX = "test_open_close_index";
 
-    @Override
-    protected Settings featureFlagSettings() {
-        return Settings.builder().put(super.featureFlagSettings()).put(FeatureFlags.READER_WRITER_SPLIT_EXPERIMENTAL, Boolean.TRUE).build();
-    }
-
     public void testIndexCloseAndOpen() throws Exception {
         internalCluster().startClusterManagerOnlyNode();
         internalCluster().startDataOnlyNodes(2);

server/src/internalClusterTest/java/org/opensearch/indices/replication/SearchReplicaReplicationAndRecoveryIT.java

@@ -21,7 +21,6 @@
 import org.opensearch.cluster.metadata.IndexMetadata;
 import org.opensearch.cluster.routing.RecoverySource;
 import org.opensearch.common.settings.Settings;
-import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.index.SegmentReplicationPerGroupStats;
 import org.opensearch.index.SegmentReplicationShardStats;
 import org.opensearch.indices.recovery.RecoveryState;
@@ -72,11 +71,6 @@ public Settings indexSettings() {
             .build();
     }
 
-    @Override
-    protected Settings featureFlagSettings() {
-        return Settings.builder().put(super.featureFlagSettings()).put(FeatureFlags.READER_WRITER_SPLIT_EXPERIMENTAL, true).build();
-    }
-
     public void testReplication() throws Exception {
         internalCluster().startClusterManagerOnlyNode();
         final String primary = internalCluster().startDataOnlyNode();

server/src/internalClusterTest/java/org/opensearch/indices/replication/SearchReplicaRestoreIT.java

@@ -16,7 +16,6 @@
 import org.opensearch.cluster.metadata.IndexMetadata;
 import org.opensearch.cluster.metadata.Metadata;
 import org.opensearch.common.settings.Settings;
-import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.index.query.QueryBuilders;
 import org.opensearch.indices.replication.common.ReplicationType;
 import org.opensearch.remotestore.RemoteSnapshotIT;
@@ -39,11 +38,6 @@ public class SearchReplicaRestoreIT extends RemoteSnapshotIT {
     private static final String FS_REPOSITORY_TYPE = "fs";
     private static final int DOC_COUNT = 10;
 
-    @Override
-    protected Settings featureFlagSettings() {
-        return Settings.builder().put(super.featureFlagSettings()).put(FeatureFlags.READER_WRITER_SPLIT_EXPERIMENTAL, true).build();
-    }
-
     public void testSearchReplicaRestore_WhenSnapshotOnSegRep_RestoreOnDocRepWithSearchReplica() throws Exception {
         bootstrapIndexWithOutSearchReplicas(ReplicationType.SEGMENT);
         createRepoAndSnapshot(REPOSITORY_NAME, FS_REPOSITORY_TYPE, SNAPSHOT_NAME, INDEX_NAME);

server/src/internalClusterTest/java/org/opensearch/indices/settings/SearchOnlyReplicaFeatureFlagIT.java

@@ -18,7 +18,6 @@
 import org.opensearch.cluster.routing.Preference;
 import org.opensearch.cluster.routing.ShardRouting;
 import org.opensearch.common.settings.Settings;
-import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.index.query.QueryBuilders;
 import org.opensearch.indices.replication.common.ReplicationType;
 import org.opensearch.remotestore.RemoteStoreBaseIntegTestCase;
@@ -40,11 +39,6 @@ public class SearchOnlyReplicaIT extends RemoteStoreBaseIntegTestCase {
 
     private static final String TEST_INDEX = "test_index";
 
-    @Override
-    protected Settings featureFlagSettings() {
-        return Settings.builder().put(super.featureFlagSettings()).put(FeatureFlags.READER_WRITER_SPLIT_EXPERIMENTAL, Boolean.TRUE).build();
-    }
-
     private final String expectedFailureMessage = "To set index.number_of_search_replicas, index.replication.type must be set to SEGMENT";
 
     @Override