Merge branch 'main' into updateGoogleJimfs

Signed-off-by: Stephen Crawford <65832608+scrawfor99@users.noreply.github.com>

.gitignore

@@ -17,6 +17,9 @@ out/
 benchmarks/src/main/generated/*
 benchmarks/bin/*
 benchmarks/build-eclipse-default/*
+server/bin/*
+server/build-eclipse-default/*
+test/framework/build-eclipse-default/*
 
 # eclipse files
 .project
@@ -61,4 +64,4 @@ testfixtures_shared/
 .ci/jobs/
 
 # build files generated
-doc-tools/missing-doclet/bin/
+doc-tools/missing-doclet/bin/

CHANGELOG.md

@@ -144,6 +144,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Bump `com.networknt:json-schema-validator` from 1.0.85 to 1.0.86 ([#8573](https://github.com/opensearch-project/OpenSearch/pull/8573))
 - Bump `com.google.cloud:google-cloud-core-http` from 2.17.0 to 2.21.0 ([#8586](https://github.com/opensearch-project/OpenSearch/pull/8586))
 - Bump `com.google.jimfs:jimfs` from 1.2 to 1.3.0 ([#8585](https://github.com/opensearch-project/OpenSearch/pull/8585))
+- Bump `org.apache.logging.log4j:log4j-core` from 2.17.1 to 2.20.0 ([#8307](https://github.com/opensearch-project/OpenSearch/pull/8307))
 
 ### Changed
 - Replace jboss-annotations-api_1.2_spec with jakarta.annotation-api ([#7836](https://github.com/opensearch-project/OpenSearch/pull/7836))

buildSrc/src/testKit/thirdPartyAudit/sample_jars/build.gradle

@@ -16,7 +16,7 @@ repositories {
   mavenCentral()
 }
 dependencies {
-  implementation 'org.apache.logging.log4j:log4j-core:2.20.0'
+  implementation "org.apache.logging.log4j:log4j-core:2.20.0"
 }
 
 ["0.0.1", "0.0.2"].forEach { v ->

buildSrc/version.properties

@@ -14,8 +14,7 @@ jackson_databind  = 2.15.2
 snakeyaml         = 2.0
 icu4j             = 70.1
 supercsv          = 2.4.0
-# Update to 2.17.2+ is breaking OpenSearchJsonLayout (see https://issues.apache.org/jira/browse/LOG4J2-3562)
-log4j             = 2.17.1
+log4j             = 2.20.0
 slf4j             = 1.7.36
 asm               = 9.5
 jettison          = 1.5.4

libs/core/licenses/log4j-api-2.20.0.jar.sha1

@@ -0,0 +1 @@
+1fe6082e660daf07c689a89c94dc0f49c26b44bb

plugins/discovery-azure-classic/licenses/log4j-1.2-api-2.20.0.jar.sha1

@@ -0,0 +1 @@
+689151374756cb809cb029f2501015bdc7733179

plugins/discovery-ec2/licenses/log4j-1.2-api-2.20.0.jar.sha1

@@ -0,0 +1 @@
+689151374756cb809cb029f2501015bdc7733179

plugins/discovery-gce/licenses/log4j-1.2-api-2.20.0.jar.sha1

@@ -0,0 +1 @@
+689151374756cb809cb029f2501015bdc7733179

plugins/repository-gcs/licenses/log4j-1.2-api-2.20.0.jar.sha1

@@ -0,0 +1 @@
+689151374756cb809cb029f2501015bdc7733179

plugins/repository-hdfs/licenses/log4j-slf4j-impl-2.20.0.jar.sha1

@@ -0,0 +1 @@
+7ab4f082fd162f60afcaf2b8744a3d959feab3e8

plugins/repository-s3/licenses/log4j-1.2-api-2.20.0.jar.sha1

@@ -0,0 +1 @@
+689151374756cb809cb029f2501015bdc7733179

qa/rolling-upgrade/src/test/java/org/opensearch/upgrades/IndexingIT.java

@@ -247,6 +247,7 @@ public void testIndexing() throws IOException, ParseException {
      *
      * @throws Exception
      */
+    @AwaitsFix(bugUrl = "https://github.com/opensearch-project/OpenSearch/issues/8322")
     public void testIndexingWithSegRep() throws Exception {
         if (UPGRADE_FROM_VERSION.before(Version.V_2_4_0)) {
             logger.info("--> Skip test for version {} where segment replication feature is not available", UPGRADE_FROM_VERSION);

qa/wildfly/build.gradle

@@ -58,11 +58,10 @@ dependencies {
   api "com.fasterxml.jackson.jakarta.rs:jackson-jakarta-rs-base:${versions.jackson}"
   api "com.fasterxml.jackson.jakarta.rs:jackson-jakarta-rs-json-provider:${versions.jackson}"
   api "com.github.fge:json-patch:1.9"
-  api "org.apache.logging.log4j:log4j-api:${versions.log4j}"
-  api "org.apache.logging.log4j:log4j-core:${versions.log4j}"
   api(project(path: ':client:rest-high-level')) {
     exclude module: 'jakarta.annotation-api'
   }
+  testImplementation "org.apache.logging.log4j:log4j-slf4j-impl:${versions.log4j}"
   testImplementation(project(':test:framework')) {
     exclude module: 'jakarta.annotation-api'
   }

server/build.gradle

@@ -146,6 +146,7 @@ dependencies {
   api "org.apache.logging.log4j:log4j-api:${versions.log4j}"
   api "org.apache.logging.log4j:log4j-jul:${versions.log4j}"
   api "org.apache.logging.log4j:log4j-core:${versions.log4j}", optional
+  annotationProcessor "org.apache.logging.log4j:log4j-core:${versions.log4j}"
 
   // jna
   api "net.java.dev.jna:jna:${versions.jna}"
@@ -175,6 +176,10 @@ tasks.withType(JavaCompile).configureEach {
   options.compilerArgs -= '-Xlint:unchecked'
 }
 
+compileJava {
+  options.compilerArgs += ['-processor', 'org.apache.logging.log4j.core.config.plugins.processor.PluginProcessor']
+}
+
 tasks.named("internalClusterTest").configure {
   // TODO: these run faster with C2 only because they run for so, so long
   jvmArgs -= '-XX:TieredStopAtLevel=1'

server/licenses/log4j-api-2.20.0.jar.sha1

@@ -0,0 +1 @@
+1fe6082e660daf07c689a89c94dc0f49c26b44bb

server/licenses/log4j-core-2.20.0.jar.sha1

@@ -0,0 +1 @@
+eb2a9a47b1396e00b5eee1264296729a70565cc0

server/licenses/log4j-jul-2.20.0.jar.sha1

@@ -0,0 +1 @@
+8170e6118eac1ab332046c179718a0f107f688e1

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

@@ -19,12 +19,15 @@
 import org.opensearch.common.Nullable;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.lease.Releasable;
+import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.index.Index;
 import org.opensearch.index.IndexModule;
 import org.opensearch.index.IndexService;
 import org.opensearch.index.SegmentReplicationPerGroupStats;
 import org.opensearch.index.SegmentReplicationShardStats;
+import org.opensearch.index.engine.Engine;
 import org.opensearch.index.shard.IndexShard;
+import org.opensearch.index.shard.ShardId;
 import org.opensearch.index.store.Store;
 import org.opensearch.index.store.StoreFileMetadata;
 import org.opensearch.indices.IndicesService;
@@ -158,13 +161,16 @@ protected void verifyStoreContent() throws Exception {
                     final String indexName = primaryRouting.getIndexName();
                     final List<ShardRouting> replicaRouting = shardRoutingTable.replicaShards();
                     final IndexShard primaryShard = getIndexShard(clusterState, primaryRouting, indexName);
+                    final int primaryDocCount = getDocCountFromShard(primaryShard);
                     final Map<String, StoreFileMetadata> primarySegmentMetadata = primaryShard.getSegmentMetadataMap();
                     for (ShardRouting replica : replicaRouting) {
                         IndexShard replicaShard = getIndexShard(clusterState, replica, indexName);
                         final Store.RecoveryDiff recoveryDiff = Store.segmentReplicationDiff(
                             primarySegmentMetadata,
                             replicaShard.getSegmentMetadataMap()
                         );
+                        final int replicaDocCount = getDocCountFromShard(replicaShard);
+                        assertEquals("Doc counts should match", primaryDocCount, replicaDocCount);
                         if (recoveryDiff.missing.isEmpty() == false || recoveryDiff.different.isEmpty() == false) {
                             fail(
                                 "Expected no missing or different segments between primary and replica but diff was missing: "
@@ -185,10 +191,30 @@ protected void verifyStoreContent() throws Exception {
         }, 1, TimeUnit.MINUTES);
     }
 
+    private int getDocCountFromShard(IndexShard shard) {
+        try (final Engine.Searcher searcher = shard.acquireSearcher("test")) {
+            return searcher.getDirectoryReader().numDocs();
+        }
+    }
+
     private IndexShard getIndexShard(ClusterState state, ShardRouting routing, String indexName) {
-        return getIndexShard(state.nodes().get(routing.currentNodeId()).getName(), indexName);
+        return getIndexShard(state.nodes().get(routing.currentNodeId()).getName(), routing.shardId(), indexName);
     }
 
+    /**
+     * Fetch IndexShard by shardId, multiple shards per node allowed.
+     */
+    protected IndexShard getIndexShard(String node, ShardId shardId, String indexName) {
+        final Index index = resolveIndex(indexName);
+        IndicesService indicesService = internalCluster().getInstance(IndicesService.class, node);
+        IndexService indexService = indicesService.indexServiceSafe(index);
+        final Optional<Integer> id = indexService.shardIds().stream().filter(sid -> sid == shardId.id()).findFirst();
+        return indexService.getShard(id.get());
+    }
+
+    /**
+     * Fetch IndexShard, assumes only a single shard per node.
+     */
     protected IndexShard getIndexShard(String node, String indexName) {
         final Index index = resolveIndex(indexName);
         IndicesService indicesService = internalCluster().getInstance(IndicesService.class, node);
@@ -197,6 +223,11 @@ protected IndexShard getIndexShard(String node, String indexName) {
         return indexService.getShard(shardId.get());
     }
 
+    protected boolean segmentReplicationWithRemoteEnabled() {
+        return IndexMetadata.INDEX_REMOTE_STORE_ENABLED_SETTING.get(indexSettings()).booleanValue()
+            && "true".equalsIgnoreCase(featureFlagSettings().get(FeatureFlags.SEGMENT_REPLICATION_EXPERIMENTAL));
+    }
+
     protected Releasable blockReplication(List<String> nodes, CountDownLatch latch) {
         CountDownLatch pauseReplicationLatch = new CountDownLatch(nodes.size());
         for (String node : nodes) {
@@ -206,7 +237,11 @@ protected Releasable blockReplication(List<String> nodes, CountDownLatch latch)
                 node
             ));
             mockTargetTransportService.addSendBehavior((connection, requestId, action, request, options) -> {
-                if (action.equals(SegmentReplicationSourceService.Actions.UPDATE_VISIBLE_CHECKPOINT)) {
+                String actionToWaitFor = SegmentReplicationSourceService.Actions.GET_SEGMENT_FILES;
+                if (segmentReplicationWithRemoteEnabled()) {
+                    actionToWaitFor = SegmentReplicationSourceService.Actions.UPDATE_VISIBLE_CHECKPOINT;
+                }
+                if (action.equals(actionToWaitFor)) {
                     try {
                         latch.countDown();
                         pauseReplicationLatch.await();

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

@@ -49,7 +49,6 @@
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.common.lease.Releasable;
-import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.index.IndexModule;
 import org.opensearch.index.SegmentReplicationPerGroupStats;
 import org.opensearch.index.SegmentReplicationPressureService;
@@ -1324,9 +1323,4 @@ public void testPrimaryReceivesDocsDuringReplicaRecovery() throws Exception {
         ensureGreen(INDEX_NAME);
         waitForSearchableDocs(2, nodes);
     }
-
-    private boolean segmentReplicationWithRemoteEnabled() {
-        return IndexMetadata.INDEX_REMOTE_STORE_ENABLED_SETTING.get(indexSettings()).booleanValue()
-            && "true".equalsIgnoreCase(featureFlagSettings().get(FeatureFlags.SEGMENT_REPLICATION_EXPERIMENTAL));
-    }
 }

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

@@ -0,0 +1,88 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.indices.replication;
+
+import org.junit.Before;
+import org.opensearch.action.admin.indices.delete.DeleteIndexRequest;
+import org.opensearch.cluster.metadata.IndexMetadata;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.indices.replication.common.ReplicationType;
+import org.opensearch.test.OpenSearchIntegTestCase;
+
+@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.SUITE, minNumDataNodes = 2)
+public class SegmentReplicationSuiteIT extends SegmentReplicationBaseIT {
+
+    @Before
+    public void setup() {
+        internalCluster().startClusterManagerOnlyNode();
+        createIndex(INDEX_NAME);
+    }
+
+    @Override
+    public Settings indexSettings() {
+        final Settings.Builder builder = Settings.builder()
+            .put(super.indexSettings())
+            // reset shard & replica count to random values set by OpenSearchIntegTestCase.
+            .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, numberOfShards())
+            .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, numberOfReplicas())
+            .put(IndexMetadata.SETTING_REPLICATION_TYPE, ReplicationType.SEGMENT);
+
+        // TODO: Randomly enable remote store on these tests.
+        return builder.build();
+    }
+
+    public void testBasicReplication() throws Exception {
+        final int docCount = scaledRandomIntBetween(10, 200);
+        for (int i = 0; i < docCount; i++) {
+            client().prepareIndex(INDEX_NAME).setId(Integer.toString(i)).setSource("field", "value" + i).execute().get();
+        }
+        refresh();
+        ensureGreen(INDEX_NAME);
+        verifyStoreContent();
+    }
+
+    public void testDropRandomNodeDuringReplication() throws Exception {
+        internalCluster().ensureAtLeastNumDataNodes(2);
+        internalCluster().startClusterManagerOnlyNodes(1);
+
+        final int docCount = scaledRandomIntBetween(10, 200);
+        for (int i = 0; i < docCount; i++) {
+            client().prepareIndex(INDEX_NAME).setId(Integer.toString(i)).setSource("field", "value" + i).execute().get();
+        }
+        refresh();
+
+        internalCluster().restartRandomDataNode();
+
+        ensureYellow(INDEX_NAME);
+        client().prepareIndex(INDEX_NAME).setId(Integer.toString(docCount)).setSource("field", "value" + docCount).execute().get();
+        internalCluster().startDataOnlyNode();
+        client().admin().indices().delete(new DeleteIndexRequest(INDEX_NAME)).actionGet();
+    }
+
+    public void testDeleteIndexWhileReplicating() throws Exception {
+        internalCluster().startClusterManagerOnlyNode();
+        final int docCount = scaledRandomIntBetween(10, 200);
+        for (int i = 0; i < docCount; i++) {
+            client().prepareIndex(INDEX_NAME).setId(Integer.toString(i)).setSource("field", "value" + i).execute().get();
+        }
+        refresh(INDEX_NAME);
+        client().admin().indices().delete(new DeleteIndexRequest(INDEX_NAME)).actionGet();
+    }
+
+    public void testFullRestartDuringReplication() throws Exception {
+        internalCluster().startNode();
+        final int docCount = scaledRandomIntBetween(10, 200);
+        for (int i = 0; i < docCount; i++) {
+            client().prepareIndex(INDEX_NAME).setId(Integer.toString(i)).setSource("field", "value" + i).execute().get();
+        }
+        refresh(INDEX_NAME);
+        internalCluster().fullRestart();
+        ensureGreen(INDEX_NAME);
+    }
+}