Merge branch 'opensearch-project:main' into segrep-throttle

.github/workflows/detect-breaking-change.yml

@@ -1,5 +1,7 @@
 name: "Detect Breaking Changes"
-on: [push, pull_request]
+on:
+  pull_request
+
 jobs:
   detect-breaking-change:
     runs-on: ubuntu-latest
@@ -11,6 +13,7 @@ jobs:
         java-version: 21
     - uses: gradle/gradle-build-action@v3
       with:
+        cache-disabled: true
         arguments: japicmp
         gradle-version: 8.7
         build-root-directory: server

CHANGELOG.md

@@ -109,6 +109,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Allow setting KEYSTORE_PASSWORD through env variable ([#12865](https://github.com/opensearch-project/OpenSearch/pull/12865))
 - [Concurrent Segment Search] Perform buildAggregation concurrently and support Composite Aggregations ([#12697](https://github.com/opensearch-project/OpenSearch/pull/12697))
 - [Concurrent Segment Search] Disable concurrent segment search for system indices and throttled requests ([#12954](https://github.com/opensearch-project/OpenSearch/pull/12954))
+- Derived fields support to derive field values at query time without indexing ([#12569](https://github.com/opensearch-project/OpenSearch/pull/12569))
 - Detect breaking changes on pull requests ([#9044](https://github.com/opensearch-project/OpenSearch/pull/9044))
 - Add cluster primary balance contraint for rebalancing with buffer ([#12656](https://github.com/opensearch-project/OpenSearch/pull/12656))
 
@@ -117,7 +118,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Bump `asm` from 9.6 to 9.7 ([#12908](https://github.com/opensearch-project/OpenSearch/pull/12908))
 - Bump `net.minidev:json-smart` from 2.5.0 to 2.5.1 ([#12893](https://github.com/opensearch-project/OpenSearch/pull/12893))
 - Bump `netty` from 4.1.107.Final to 4.1.108.Final ([#12924](https://github.com/opensearch-project/OpenSearch/pull/12924))
-- Bump `commons-io:commons-io` from 2.15.1 to 2.16.0 ([#12996](https://github.com/opensearch-project/OpenSearch/pull/12996))
+- Bump `commons-io:commons-io` from 2.15.1 to 2.16.0 ([#12996](https://github.com/opensearch-project/OpenSearch/pull/12996), [#12998](https://github.com/opensearch-project/OpenSearch/pull/12998), [#12999](https://github.com/opensearch-project/OpenSearch/pull/12999))
 
 ### Changed
 - [BWC and API enforcement] Enforcing the presence of API annotations at build time ([#12872](https://github.com/opensearch-project/OpenSearch/pull/12872))

plugins/discovery-azure-classic/build.gradle

@@ -53,7 +53,7 @@ dependencies {
   api "org.apache.logging.log4j:log4j-1.2-api:${versions.log4j}"
   api "commons-codec:commons-codec:${versions.commonscodec}"
   api "commons-lang:commons-lang:2.6"
-  api "commons-io:commons-io:2.15.1"
+  api "commons-io:commons-io:2.16.0"
   api 'javax.mail:mail:1.4.7'
   api 'javax.inject:javax.inject:1'
   api "com.sun.jersey:jersey-client:${versions.jersey}"

plugins/discovery-azure-classic/licenses/commons-io-2.16.0.jar.sha1

@@ -0,0 +1 @@
+27875a7935f1ddcc13267eb6fae1f719e0409572

plugins/ingest-attachment/build.gradle

@@ -57,7 +57,7 @@ dependencies {
   runtimeOnly "com.google.guava:guava:${versions.guava}"
   // Other dependencies
   api 'org.tukaani:xz:1.9'
-  api 'commons-io:commons-io:2.15.1'
+  api 'commons-io:commons-io:2.16.0'
   api "org.slf4j:slf4j-api:${versions.slf4j}"
 
   // character set detection

plugins/ingest-attachment/licenses/commons-io-2.16.0.jar.sha1

@@ -0,0 +1 @@
+27875a7935f1ddcc13267eb6fae1f719e0409572

server/src/internalClusterTest/java/org/opensearch/index/shard/IndexShardIT.java

@@ -715,7 +715,8 @@ public static final IndexShard newIndexShard(
             nodeId,
             null,
             DefaultRemoteStoreSettings.INSTANCE,
-            false
+            false,
+            IndexShardTestUtils.getFakeDiscoveryNodes(initializingShardRouting)
         );
     }
 

server/src/internalClusterTest/java/org/opensearch/remotemigration/MigrationBaseTestCase.java

@@ -8,17 +8,29 @@
 
 package org.opensearch.remotemigration;
 
+import org.opensearch.action.DocWriteResponse;
 import org.opensearch.action.admin.cluster.repositories.get.GetRepositoriesRequest;
 import org.opensearch.action.admin.cluster.repositories.get.GetRepositoriesResponse;
+import org.opensearch.action.bulk.BulkRequest;
+import org.opensearch.action.bulk.BulkResponse;
+import org.opensearch.action.delete.DeleteResponse;
+import org.opensearch.action.index.IndexRequest;
+import org.opensearch.action.index.IndexResponse;
 import org.opensearch.cluster.metadata.RepositoryMetadata;
+import org.opensearch.common.UUIDs;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.repositories.fs.ReloadableFsRepository;
 import org.opensearch.test.OpenSearchIntegTestCase;
 
 import java.nio.file.Path;
+import java.util.List;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 
+import static org.opensearch.node.remotestore.RemoteStoreNodeService.MIGRATION_DIRECTION_SETTING;
+import static org.opensearch.node.remotestore.RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING;
 import static org.opensearch.repositories.fs.ReloadableFsRepository.REPOSITORIES_FAILRATE_SETTING;
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
 
@@ -28,8 +40,16 @@ public class MigrationBaseTestCase extends OpenSearchIntegTestCase {
 
     protected Path segmentRepoPath;
     protected Path translogRepoPath;
-
     boolean addRemote = false;
+    Settings extraSettings = Settings.EMPTY;
+
+    private final List<String> documentKeys = List.of(
+        randomAlphaOfLength(5),
+        randomAlphaOfLength(5),
+        randomAlphaOfLength(5),
+        randomAlphaOfLength(5),
+        randomAlphaOfLength(5)
+    );
 
     protected Settings nodeSettings(int nodeOrdinal) {
         if (segmentRepoPath == null || translogRepoPath == null) {
@@ -40,6 +60,7 @@ protected Settings nodeSettings(int nodeOrdinal) {
             logger.info("Adding remote store node");
             return Settings.builder()
                 .put(super.nodeSettings(nodeOrdinal))
+                .put(extraSettings)
                 .put(remoteStoreClusterSettings(REPOSITORY_NAME, segmentRepoPath, REPOSITORY_2_NAME, translogRepoPath))
                 .build();
         } else {
@@ -64,4 +85,76 @@ protected void setFailRate(String repoName, int value) throws ExecutionException
             client().admin().cluster().preparePutRepository(repoName).setType(ReloadableFsRepository.TYPE).setSettings(settings).get()
         );
     }
+
+    public void initDocRepToRemoteMigration() {
+        assertTrue(
+            internalCluster().client()
+                .admin()
+                .cluster()
+                .prepareUpdateSettings()
+                .setPersistentSettings(
+                    Settings.builder()
+                        .put(REMOTE_STORE_COMPATIBILITY_MODE_SETTING.getKey(), "mixed")
+                        .put(MIGRATION_DIRECTION_SETTING.getKey(), "remote_store")
+                )
+                .get()
+                .isAcknowledged()
+        );
+    }
+
+    public BulkResponse indexBulk(String indexName, int numDocs) {
+        BulkRequest bulkRequest = new BulkRequest();
+        for (int i = 0; i < numDocs; i++) {
+            final IndexRequest request = client().prepareIndex(indexName)
+                .setId(UUIDs.randomBase64UUID())
+                .setSource(documentKeys.get(randomIntBetween(0, documentKeys.size() - 1)), randomAlphaOfLength(5))
+                .request();
+            bulkRequest.add(request);
+        }
+        return client().bulk(bulkRequest).actionGet();
+    }
+
+    private void indexSingleDoc(String indexName) {
+        IndexResponse indexResponse = client().prepareIndex(indexName).setId("id").setSource("field", "value").get();
+        assertEquals(DocWriteResponse.Result.CREATED, indexResponse.getResult());
+        DeleteResponse deleteResponse = client().prepareDelete(indexName, "id").get();
+        assertEquals(DocWriteResponse.Result.DELETED, deleteResponse.getResult());
+        client().prepareIndex(indexName).setSource("auto", true).get();
+    }
+
+    public class AsyncIndexingService {
+        private String indexName;
+        private AtomicLong indexedDocs = new AtomicLong(0);
+        private AtomicBoolean finished = new AtomicBoolean();
+        private Thread indexingThread;
+
+        AsyncIndexingService(String indexName) {
+            this.indexName = indexName;
+        }
+
+        public void startIndexing() {
+            indexingThread = getIndexingThread();
+            indexingThread.start();
+        }
+
+        public void stopIndexing() throws InterruptedException {
+            finished.set(true);
+            indexingThread.join();
+        }
+
+        public long getIndexedDocs() {
+            return indexedDocs.get();
+        }
+
+        private Thread getIndexingThread() {
+            return new Thread(() -> {
+                while (finished.get() == false) {
+                    indexSingleDoc(indexName);
+                    long currentDocCount = indexedDocs.incrementAndGet();
+                    logger.info("Completed ingestion of {} docs", currentDocCount);
+
+                }
+            });
+        }
+    }
 }