Merge branch 'main' into remote_connect

Signed-off-by: Ankit Kala <ankikala@amazon.com>

.github/workflows/poc-checklist.yml

@@ -0,0 +1,49 @@
+name: Add comment
+on:
+  issues:
+    types:
+      - labeled
+jobs:
+  add-comment:
+    if: github.event.label.name == 'poc'
+    runs-on: ubuntu-latest
+    permissions:
+      issues: write
+    steps:
+      - name: Add comment
+        uses: peter-evans/create-or-update-comment@v2
+        with:
+          issue-number: ${{ github.event.issue.number }}
+          body: |
+            ### POC Checklist:
+
+            Please go through the following checklist to ensure these items are taken into account while designing the POC.
+
+            - [ ] Supports safe upgrade paths from all supported BWC versions to the current version
+            - [ ] Supports compatibility with all plugins
+                - [ ] opensearch-alerting
+                - [ ] opensearch-anomaly-detection
+                - [ ] opensearch-asynchronous-search
+                - [ ] opensearch-cross-cluster-replication
+                - [ ] opensearch-geospatial
+                - [ ] opensearch-index-management
+                - [ ] opensearch-job-scheduler
+                - [ ] opensearch-knn
+                - [ ] opensearch-ml
+                - [ ] opensearch-notifications
+                - [ ] opensearch-notifications-core
+                - [ ] opensearch-observability
+                - [ ] opensearch-performance-analyzer
+                - [ ] opensearch-reports-scheduler
+                - [ ] opensearch-security
+                - [ ] opensearch-sql
+            - [ ] Supports lucene upgrades across minor lucene versions
+            - [ ] Supports lucene upgrades across major lucene versions
+            - [ ] Supports lucene upgrades across underlying lucene codec bumps (Eg: Lucene95Codec -> Lucene96Codec)
+            - [ ] Supports wire compatibility of OpenSearch
+            - [ ] Plan to measure performance degradation/improvement (if any)
+            - [ ] Plan to document any user facing changes introduced by this feature
+            - [ ] Ensure working and passing CI
+
+            Thank you for your contribution!
+

CHANGELOG.md

@@ -50,6 +50,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Bump `com.networknt:json-schema-validator` from 1.0.81 to 1.0.83 ([7933](https://github.com/opensearch-project/OpenSearch/pull/7933/))
 - Bump `com.google.http-client:google-http-client:1.43.2` from 1.42.0 to 1.43.2 ([7928](https://github.com/opensearch-project/OpenSearch/pull/7928)))
 - Bump `com.azure:azure-storage-blob` from 12.21.1 to 12.22.2 ([7930](https://github.com/opensearch-project/OpenSearch/pull/7930))
+- Bump `io.projectreactor:reactor-core` from 3.5.1 to 3.5.6 in /plugins/repository-azure ([#8016](https://github.com/opensearch-project/OpenSearch/pull/8016))
+- Bump `com.google.guava:guava` from 32.0.0-jre to 32.0.1-jre in /distribution/tools/upgrade-cli ([#8011](https://github.com/opensearch-project/OpenSearch/pull/8011))
+- Bump `com.google.guava:guava` from 32.0.0-jre to 32.0.1-jre in /distribution/tools/plugin-cli ([#8012](https://github.com/opensearch-project/OpenSearch/pull/8012))
 
 ### Changed
 - [CCR] Add getHistoryOperationsFromTranslog method to fetch the history snapshot from translogs ([#3948](https://github.com/opensearch-project/OpenSearch/pull/3948))
@@ -81,6 +84,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Remove LegacyESVersion.V_7_10_ Constants ([#5018](https://github.com/opensearch-project/OpenSearch/pull/5018))
 - Remove Version.V_1_ Constants ([#5021](https://github.com/opensearch-project/OpenSearch/pull/5021))
 - Remove custom Map, List and Set collection classes ([#6871](https://github.com/opensearch-project/OpenSearch/pull/6871))
+- Remove COMPRESSOR variable from CompressorFactory and use DEFLATE_COMPRESSOR instead ([7907](https://github.com/opensearch-project/OpenSearch/pull/7907))
 
 ### Fixed
 - Fix 'org.apache.hc.core5.http.ParseException: Invalid protocol version' under JDK 16+ ([#4827](https://github.com/opensearch-project/OpenSearch/pull/4827))
@@ -98,13 +102,16 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add Remote store as a segment replication source ([#7653](https://github.com/opensearch-project/OpenSearch/pull/7653))
 - Add descending order search optimization through reverse segment read. ([#7967](https://github.com/opensearch-project/OpenSearch/pull/7967))
 - Make remote cluster connection setup in async ([#8038](https://github.com/opensearch-project/OpenSearch/pull/8038))
+- Update components of segrep backpressure to support remote store. ([#8020](https://github.com/opensearch-project/OpenSearch/pull/8020))
 
 ### Dependencies
 - Bump `jackson` from 2.15.1 to 2.15.2 ([#7897](https://github.com/opensearch-project/OpenSearch/pull/7897))
 - Bump `netty` from 4.1.91.Final to 4.1.93.Final ([#7901](https://github.com/opensearch-project/OpenSearch/pull/7901))
 - Bump `com.netflix.nebula:gradle-extra-configurations-plugin` from 9.0.0 to 10.0.0 in /buildSrc ([#7068](https://github.com/opensearch-project/OpenSearch/pull/7068))
 - Bump `com.google.guava:guava` from 32.0.0-jre to 32.0.1-jre (#8009)
-- Bump `commons-io:commons-io` from 2.12.0 to 2.13.0 (#8014, #8013)
+- Bump `commons-io:commons-io` from 2.12.0 to 2.13.0 (#8014, #8013, #8010)
+- Bump `com.diffplug.spotless` from 6.18.0 to 6.19.0 (#8007)
+- Bump `org.jruby.joni:joni` from 2.1.48 to 2.2.1 (#8015)
 
 ### Changed
 - Replace jboss-annotations-api_1.2_spec with jakarta.annotation-api ([#7836](https://github.com/opensearch-project/OpenSearch/pull/7836))
@@ -123,4 +130,4 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Security
 
 [Unreleased 3.0]: https://github.com/opensearch-project/OpenSearch/compare/2.x...HEAD
-[Unreleased 2.x]: https://github.com/opensearch-project/OpenSearch/compare/2.8...2.x
+[Unreleased 2.x]: https://github.com/opensearch-project/OpenSearch/compare/2.8...2.x

build.gradle

@@ -53,7 +53,7 @@ plugins {
   id 'lifecycle-base'
   id 'opensearch.docker-support'
   id 'opensearch.global-build-info'
-  id "com.diffplug.spotless" version "6.18.0" apply false
+  id "com.diffplug.spotless" version "6.19.0" apply false
   id "org.gradle.test-retry" version "1.5.3" apply false
   id "test-report-aggregation"
   id 'jacoco-report-aggregation'

distribution/tools/plugin-cli/build.gradle

@@ -39,7 +39,7 @@ dependencies {
   api "org.bouncycastle:bc-fips:1.0.2.3"
   testImplementation project(":test:framework")
   testImplementation 'com.google.jimfs:jimfs:1.2'
-  testRuntimeOnly 'com.google.guava:guava:32.0.0-jre'
+  testRuntimeOnly 'com.google.guava:guava:32.0.1-jre'
 
   implementation 'org.apache.commons:commons-compress:1.23.0'
 }

distribution/tools/upgrade-cli/build.gradle

@@ -19,7 +19,7 @@ dependencies {
   implementation "com.fasterxml.jackson.core:jackson-annotations:${versions.jackson}"
   testImplementation project(":test:framework")
   testImplementation 'com.google.jimfs:jimfs:1.2'
-  testRuntimeOnly 'com.google.guava:guava:32.0.0-jre'
+  testRuntimeOnly 'com.google.guava:guava:32.0.1-jre'
 }
 
 tasks.named("dependencyLicenses").configure {

libs/grok/build.gradle

@@ -29,7 +29,7 @@
  */
 
 dependencies {
-  api 'org.jruby.joni:joni:2.1.48'
+  api 'org.jruby.joni:joni:2.2.1'
   // joni dependencies:
   api 'org.jruby.jcodings:jcodings:1.0.58'
 

libs/grok/licenses/joni-2.2.1.jar.sha1

@@ -0,0 +1 @@
+23d2f2eff7fa0cda465d86ec9d8bab53e496d9e6

libs/grok/src/main/java/org/opensearch/grok/Grok.java

@@ -214,8 +214,8 @@ private String groupMatch(String name, Region region, String pattern) {
                 name.getBytes(StandardCharsets.UTF_8).length,
                 region
             );
-            int begin = region.beg[number];
-            int end = region.end[number];
+            int begin = region.getBeg(number);
+            int end = region.getEnd(number);
             return new String(pattern.getBytes(StandardCharsets.UTF_8), begin, end - begin, StandardCharsets.UTF_8);
         } catch (StringIndexOutOfBoundsException e) {
             return null;
@@ -270,7 +270,12 @@ protected String toRegex(String grokPattern) {
                 grokPart = String.format(Locale.US, "(?<%s>%s)", patternName + "_" + result, pattern);
             }
             String start = new String(grokPatternBytes, 0, result, StandardCharsets.UTF_8);
-            String rest = new String(grokPatternBytes, region.end[0], grokPatternBytes.length - region.end[0], StandardCharsets.UTF_8);
+            String rest = new String(
+                grokPatternBytes,
+                region.getEnd(0),
+                grokPatternBytes.length - region.getEnd(0),
+                StandardCharsets.UTF_8
+            );
             grokPattern = grokPart + rest;
             res.append(start);
         }

libs/grok/src/main/java/org/opensearch/grok/GrokCaptureType.java

@@ -105,9 +105,9 @@ protected final GrokCaptureExtracter rawExtracter(int[] backRefs, Consumer<? sup
             @Override
             void extract(byte[] utf8Bytes, int offset, Region region) {
                 for (int number : backRefs) {
-                    if (region.beg[number] >= 0) {
-                        int matchOffset = offset + region.beg[number];
-                        int matchLength = region.end[number] - region.beg[number];
+                    if (region.getBeg(number) >= 0) {
+                        int matchOffset = offset + region.getBeg(number);
+                        int matchLength = region.getEnd(number) - region.getBeg(number);
                         emit.accept(new String(utf8Bytes, matchOffset, matchLength, StandardCharsets.UTF_8));
                         return; // Capture only the first value.
                     }

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.12.0'
+  api 'commons-io:commons-io:2.13.0'
   api "org.slf4j:slf4j-api:${versions.slf4j}"
 
   // character set detection

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

@@ -0,0 +1 @@
+8bb2bc9b4df17e2411533a0708a69f983bf5e83b

plugins/repository-azure/build.gradle

@@ -57,7 +57,7 @@ dependencies {
   implementation project(':modules:transport-netty4')
   api 'com.azure:azure-storage-blob:12.22.2'
   api 'org.reactivestreams:reactive-streams:1.0.4'
-  api 'io.projectreactor:reactor-core:3.5.1'
+  api 'io.projectreactor:reactor-core:3.5.6'
   api 'io.projectreactor.netty:reactor-netty:1.1.7'
   api 'io.projectreactor.netty:reactor-netty-core:1.1.7'
   api 'io.projectreactor.netty:reactor-netty-http:1.1.7'
@@ -170,6 +170,7 @@ thirdPartyAudit {
     'org.slf4j.impl.StaticMarkerBinder',
     'reactor.blockhound.BlockHound$Builder',
     'reactor.blockhound.integration.BlockHoundIntegration',
+    'io.micrometer.context.ThreadLocalAccessor',
     'io.micrometer.common.KeyValue',
     'io.micrometer.common.KeyValues',
     'io.micrometer.common.docs.KeyName',

plugins/repository-azure/licenses/reactor-core-3.5.6.jar.sha1

@@ -0,0 +1 @@
+027fdc551537b349389176a23a192f11a7a3d7de

qa/smoke-test-http/src/test/java/org/opensearch/http/SearchRestCancellationIT.java

@@ -83,6 +83,7 @@
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertNoFailures;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.opensearch.test.hamcrest.OpenSearchAssertions.awaitLatch;
 
 public class SearchRestCancellationIT extends HttpSmokeTestCase {
 
@@ -183,6 +184,7 @@ public void onFailure(Exception exception) {
             }
         });
 
+        latch.await(2, TimeUnit.SECONDS);
         awaitForBlock(plugins);
         cancellable.cancel();
         ensureSearchTaskIsCancelled(searchAction, nodeIdToName::get);

server/src/internalClusterTest/java/org/opensearch/index/SegmentReplicationPressureIT.java

@@ -123,9 +123,9 @@ public void testWritesRejected() throws Exception {
             assertEquals(perGroupStats.getRejectedRequestCount(), 2L);
         }
         refresh(INDEX_NAME);
-        // wait for the replicas to catch up after block is released.
-        waitForSearchableDocs(totalDocs.get(), replicaNodes.toArray(new String[] {}));
 
+        // wait for the replicas to catch up after block is released.
+        assertReplicaCheckpointUpdated(primaryShard);
         // index another doc showing there is no pressure enforced.
         indexDoc();
         refresh(INDEX_NAME);
@@ -179,7 +179,7 @@ public void testAddReplicaWhileWritesBlocked() throws Exception {
         }
         refresh(INDEX_NAME);
         // wait for the replicas to catch up after block is released.
-        waitForSearchableDocs(totalDocs.get(), replicaNodes.toArray(new String[] {}));
+        assertReplicaCheckpointUpdated(primaryShard);
 
         // index another doc showing there is no pressure enforced.
         indexDoc();
@@ -258,6 +258,10 @@ public void testFailStaleReplica() throws Exception {
     }
 
     public void testWithDocumentReplicationEnabledIndex() throws Exception {
+        assumeTrue(
+            "Can't create DocRep index with remote store enabled. Skipping.",
+            indexSettings().getAsBoolean(IndexMetadata.SETTING_REMOTE_STORE_ENABLED, false) == false
+        );
         Settings settings = Settings.builder().put(MAX_REPLICATION_TIME_SETTING.getKey(), TimeValue.timeValueMillis(500)).build();
         // Starts a primary and replica node.
         final String primaryNode = internalCluster().startNode(settings);
@@ -313,7 +317,7 @@ public void testBulkWritesRejected() throws Exception {
         }
         refresh(INDEX_NAME);
         // wait for the replicas to catch up after block is released.
-        waitForSearchableDocs(totalDocs, replicaNodes.toArray(new String[] {}));
+        assertReplicaCheckpointUpdated(primaryShard);
 
         // index another doc showing there is no pressure enforced.
         executeBulkRequest(nodes, totalDocs);

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

@@ -126,7 +126,7 @@ public void accept(final long g, final Exception e) {
             }
 
         }, null);
-        shard.close("closed", randomBoolean());
+        shard.close("closed", randomBoolean(), false);
         assertBusy(() -> assertTrue(invoked.get()));
     }
 

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

@@ -620,7 +620,7 @@ public void testShardHasMemoryBufferOnTranslogRecover() throws Throwable {
         client().prepareIndex("test").setId("1").setSource("{\"foo\" : \"bar\"}", XContentType.JSON).setRefreshPolicy(IMMEDIATE).get();
 
         CheckedFunction<DirectoryReader, DirectoryReader, IOException> wrapper = directoryReader -> directoryReader;
-        shard.close("simon says", false);
+        shard.close("simon says", false, false);
         AtomicReference<IndexShard> shardRef = new AtomicReference<>();
         List<Exception> failures = new ArrayList<>();
         IndexingOperationListener listener = new IndexingOperationListener() {
@@ -658,7 +658,7 @@ public void postDelete(ShardId shardId, Engine.Delete delete, Engine.DeleteResul
         try {
             ExceptionsHelper.rethrowAndSuppress(failures);
         } finally {
-            newShard.close("just do it", randomBoolean());
+            newShard.close("just do it", randomBoolean(), false);
         }
     }
 

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

@@ -23,6 +23,7 @@
 import org.opensearch.index.IndexModule;
 import org.opensearch.index.IndexService;
 import org.opensearch.index.SegmentReplicationPerGroupStats;
+import org.opensearch.index.SegmentReplicationShardStats;
 import org.opensearch.index.shard.IndexShard;
 import org.opensearch.index.store.Store;
 import org.opensearch.index.store.StoreFileMetadata;
@@ -38,6 +39,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -204,7 +206,7 @@ protected Releasable blockReplication(List<String> nodes, CountDownLatch latch)
                 node
             ));
             mockTargetTransportService.addSendBehavior((connection, requestId, action, request, options) -> {
-                if (action.equals(SegmentReplicationSourceService.Actions.GET_SEGMENT_FILES)) {
+                if (action.equals(SegmentReplicationSourceService.Actions.UPDATE_VISIBLE_CHECKPOINT)) {
                     try {
                         latch.countDown();
                         pauseReplicationLatch.await();
@@ -222,4 +224,13 @@ protected Releasable blockReplication(List<String> nodes, CountDownLatch latch)
         };
     }
 
+    protected void assertReplicaCheckpointUpdated(IndexShard primaryShard) throws Exception {
+        assertBusy(() -> {
+            Set<SegmentReplicationShardStats> groupStats = primaryShard.getReplicationStats();
+            assertEquals(primaryShard.indexSettings().getNumberOfReplicas(), groupStats.size());
+            for (SegmentReplicationShardStats shardStat : groupStats) {
+                assertEquals(0, shardStat.getCheckpointsBehindCount());
+            }
+        }, 30, TimeUnit.SECONDS);
+    }
 }

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

@@ -796,10 +796,6 @@ public void testReplicaHasDiffFilesThanPrimary() throws Exception {
     }
 
     public void testPressureServiceStats() throws Exception {
-        assumeFalse(
-            "Skipping the test as pressure service is not compatible with SegRep and Remote store yet.",
-            segmentReplicationWithRemoteEnabled()
-        );
         final String primaryNode = internalCluster().startNode();
         createIndex(INDEX_NAME);
         final String replicaNode = internalCluster().startNode();

server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreBaseIntegTestCase.java

@@ -18,7 +18,13 @@
 import org.opensearch.indices.replication.common.ReplicationType;
 import org.opensearch.test.OpenSearchIntegTestCase;
 
+import java.io.IOException;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
 import java.nio.file.Path;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
 
@@ -27,6 +33,8 @@ public class RemoteStoreBaseIntegTestCase extends OpenSearchIntegTestCase {
     protected static final int SHARD_COUNT = 1;
     protected static final int REPLICA_COUNT = 1;
 
+    protected Path absolutePath;
+
     @Override
     protected boolean addMockInternalEngine() {
         return false;
@@ -73,7 +81,7 @@ protected Settings remoteTranslogIndexSettings(int numberOfReplicas) {
     @Before
     public void setup() {
         internalCluster().startClusterManagerOnlyNode();
-        Path absolutePath = randomRepoPath().toAbsolutePath();
+        absolutePath = randomRepoPath().toAbsolutePath();
         assertAcked(
             clusterAdmin().preparePutRepository(REPOSITORY_NAME).setType("fs").setSettings(Settings.builder().put("location", absolutePath))
         );
@@ -84,4 +92,22 @@ public void teardown() {
         assertAcked(clusterAdmin().prepareDeleteRepository(REPOSITORY_NAME));
     }
 
+    public int getFileCount(Path path) throws Exception {
+        final AtomicInteger filesExisting = new AtomicInteger(0);
+        Files.walkFileTree(path, new SimpleFileVisitor<>() {
+            @Override
+            public FileVisitResult postVisitDirectory(Path dir, IOException impossible) throws IOException {
+                return FileVisitResult.CONTINUE;
+            }
+
+            @Override
+            public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {
+                filesExisting.incrementAndGet();
+                return FileVisitResult.CONTINUE;
+            }
+        });
+
+        return filesExisting.get();
+    }
+
 }