Fix ref count handling in Engine.failEngine (elastic#48639) (elastic#48646)

We can run into an already closed store here and hence
throw on trying to increment the ref count => moving to
the guarded ref count increment

closes elastic#48625

Author: original-brownbear

server/src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -1157,7 +1157,6 @@ public void failEngine(String reason, @Nullable Exception failure) {
             maybeDie(reason, failure);
         }
         if (failEngineLock.tryLock()) {
-            store.incRef();
             try {
                 if (failedEngine.get() != null) {
                     logger.warn(() ->
@@ -1179,11 +1178,19 @@ public void failEngine(String reason, @Nullable Exception failure) {
                     // on the same node that we don't see the corrupted marker file when
                     // the shard is initializing
                     if (Lucene.isCorruptionException(failure)) {
-                        try {
-                            store.markStoreCorrupted(new IOException("failed engine (reason: [" + reason + "])",
-                                ExceptionsHelper.unwrapCorruption(failure)));
-                        } catch (IOException e) {
-                            logger.warn("Couldn't mark store corrupted", e);
+                        if (store.tryIncRef()) {
+                            try {
+                                store.markStoreCorrupted(new IOException("failed engine (reason: [" + reason + "])",
+                                    ExceptionsHelper.unwrapCorruption(failure)));
+                            } catch (IOException e) {
+                                logger.warn("Couldn't mark store corrupted", e);
+                            } finally {
+                                store.decRef();
+                            }
+                        } else {
+                            logger.warn(() ->
+                                    new ParameterizedMessage("tried to mark store as corrupted but store is already closed. [{}]", reason),
+                                failure);
                         }
                     }
                     eventListener.onFailedEngine(reason, failure);
@@ -1192,8 +1199,6 @@ public void failEngine(String reason, @Nullable Exception failure) {
                 if (failure != null) inner.addSuppressed(failure);
                 // don't bubble up these exceptions up
                 logger.warn("failEngine threw exception", inner);
-            } finally {
-                store.decRef();
             }
         } else {
             logger.debug(() -> new ParameterizedMessage("tried to fail engine but could not acquire lock - engine should " +

server/src/test/java/org/elasticsearch/recovery/RelocationIT.java

@@ -490,7 +490,6 @@ public void testIndexAndRelocateConcurrently() throws Exception {
             docs[i] = client().prepareIndex("test", "type1", id).setSource("field1", English.intToEnglish(numDocs + i));
         }
         indexRandom(true, docs);
-        numDocs *= 2;
 
         logger.info(" --> waiting for relocation to complete");
         ensureGreen(TimeValue.timeValueSeconds(60), "test"); // move all shards to the new nodes (it waits on relocation)