Adapt CloseFollowerIndexIT for replicated closed indices (elastic#38767)

Now the test `CloseFollowerIndexIT` has been added in elastic#38702, it needs to 
be adapted for replicated closed indices.

The test closes the follower index which is lagging behind the leader index. 
When it's closed, no sanity checks are executed because it's a follower index 
(this is a consequence of elastic#38702). But with replicated closed indices, the index
 is reinitialized as a closed index with a `NoOpEngine` and such engines make 
strong assertions on the values of the maximum sequence number and the 
global checkpoint. Since the values do not match, the shards cannot be created 
and fail and the cluster health turns RED.

This commit adapts the `CloseFollowerIndexIT` test so that it wraps the 
default `UncaughtExceptionHandler` with a handler that tolerates any exception 
thrown by `ReadOnlyEngine.assertMaxSeqNoEqualsToGlobalCheckpoint()`. 
Replacing the default uncaught exception handler requires specific permissions,
 and instead of creating another gradle project it duplicates the 
`internalClusterTest` task to make it work without security manager for this 
specific test only.

Relates to elastic#33888

x-pack/plugin/ccr/build.gradle

@@ -18,18 +18,32 @@ integTest.enabled = false
 compileJava.options.compilerArgs << "-Xlint:-try"
 compileTestJava.options.compilerArgs << "-Xlint:-try"
 
+// Integration Test classes that cannot run with the security manager
+String[] noSecurityManagerITClasses = [ "**/CloseFollowerIndexIT.class" ]
+
 // Instead we create a separate task to run the
 // tests based on ESIntegTestCase
 task internalClusterTest(type: RandomizedTestingTask,
         group: JavaBasePlugin.VERIFICATION_GROUP,
         description: 'Java fantasy integration tests',
         dependsOn: unitTest.dependsOn) {
     include '**/*IT.class'
+    exclude noSecurityManagerITClasses
     systemProperty 'es.set.netty.runtime.available.processors', 'false'
 }
 check.dependsOn internalClusterTest
 internalClusterTest.mustRunAfter test
 
+task internalClusterTestNoSecurityManager(type: RandomizedTestingTask,
+        group: JavaBasePlugin.VERIFICATION_GROUP,
+        description: 'Java fantasy integration tests with no security manager',
+        dependsOn: unitTest.dependsOn) {
+    include noSecurityManagerITClasses
+    systemProperty 'es.set.netty.runtime.available.processors', 'false'
+    systemProperty 'tests.security.manager', 'false'
+}
+internalClusterTest.dependsOn internalClusterTestNoSecurityManager
+
 // add all sub-projects of the qa sub-project
 gradle.projectsEvaluated {
     project.subprojects

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/CloseFollowerIndexIT.java

@@ -11,17 +11,21 @@
 import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.block.ClusterBlock;
+import org.elasticsearch.cluster.health.ClusterHealthStatus;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.MetaDataIndexStateService;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.engine.ReadOnlyEngine;
 import org.elasticsearch.xpack.CcrIntegTestCase;
 import org.elasticsearch.xpack.core.ccr.action.PutFollowAction;
+import org.junit.After;
+import org.junit.Before;
 
-import java.util.ArrayList;
-import java.util.List;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import static java.util.Collections.singletonMap;
@@ -31,7 +35,36 @@
 
 public class CloseFollowerIndexIT extends CcrIntegTestCase {
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/pull/38767")
+    private Thread.UncaughtExceptionHandler uncaughtExceptionHandler;
+
+    @Before
+    public void wrapUncaughtExceptionHandler() {
+        uncaughtExceptionHandler = Thread.getDefaultUncaughtExceptionHandler();
+        AccessController.doPrivileged((PrivilegedAction<Void>) () -> {
+            Thread.setDefaultUncaughtExceptionHandler((t, e) -> {
+                if (t.getThreadGroup().getName().contains(getTestClass().getSimpleName())) {
+                    for (StackTraceElement element : e.getStackTrace()) {
+                        if (element.getClassName().equals(ReadOnlyEngine.class.getName())) {
+                            if (element.getMethodName().equals("assertMaxSeqNoEqualsToGlobalCheckpoint")) {
+                                return;
+                            }
+                        }
+                    }
+                }
+                uncaughtExceptionHandler.uncaughtException(t, e);
+            });
+            return null;
+        });
+    }
+
+    @After
+    public void restoreUncaughtExceptionHandler() {
+        AccessController.doPrivileged((PrivilegedAction<Void>) () -> {
+            Thread.setDefaultUncaughtExceptionHandler(uncaughtExceptionHandler);
+            return null;
+        });
+    }
+
     public void testCloseAndReopenFollowerIndex() throws Exception {
         final String leaderIndexSettings = getIndexSettings(1, 1, singletonMap(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "true"));
         assertAcked(leaderClient().admin().indices().prepareCreate("index1").setSource(leaderIndexSettings, XContentType.JSON));
@@ -67,16 +100,22 @@ public void testCloseAndReopenFollowerIndex() throws Exception {
         assertThat(response.isAcknowledged(), is(true));
 
         ClusterState clusterState = followerClient().admin().cluster().prepareState().get().getState();
-        List<ClusterBlock> blocks = new ArrayList<>(clusterState.getBlocks().indices().get("index2"));
-        assertThat(blocks.size(), equalTo(1));
-        assertThat(blocks.get(0).id(), equalTo(MetaDataIndexStateService.INDEX_CLOSED_BLOCK_ID));
+        assertThat(clusterState.metaData().index("index2").getState(), is(IndexMetaData.State.CLOSE));
+        assertThat(clusterState.getBlocks().hasIndexBlock("index2", MetaDataIndexStateService.INDEX_CLOSED_BLOCK), is(true));
+        assertThat(followerClient().admin().cluster().prepareHealth("index2").get().getStatus(), equalTo(ClusterHealthStatus.RED));
 
         isRunning.set(false);
         for (Thread thread : threads) {
             thread.join();
         }
+
         assertAcked(followerClient().admin().indices().open(new OpenIndexRequest("index2")).get());
 
+        clusterState = followerClient().admin().cluster().prepareState().get().getState();
+        assertThat(clusterState.metaData().index("index2").getState(), is(IndexMetaData.State.OPEN));
+        assertThat(clusterState.getBlocks().hasIndexBlockWithId("index2", MetaDataIndexStateService.INDEX_CLOSED_BLOCK_ID), is(false));
+        ensureFollowerGreen("index2");
+
         refresh(leaderClient(), "index1");
         SearchRequest leaderSearchRequest = new SearchRequest("index1");
         leaderSearchRequest.source().trackTotalHits(true);