Decrease store refcount on any failure to create NRTReplicationEngine (…

…#9796) * Decrease store refcount on any failure to create NRTReplicationEngine Signed-off-by: Sachin Kale <[email protected]> * Add unit tests Signed-off-by: Sachin Kale <[email protected]> * Fix spotless Signed-off-by: Sachin Kale <[email protected]> --------- Signed-off-by: Sachin Kale <[email protected]> Co-authored-by: Sachin Kale <[email protected]> (cherry picked from commit 199123d) Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
opensearch-project · Sep 7, 2023 · d0767f1 · d0767f1
1 parent d81b63b
commit d0767f1
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 2 deletions.
diff --git a/server/src/main/java/org/opensearch/index/engine/NRTReplicationEngine.java b/server/src/main/java/org/opensearch/index/engine/NRTReplicationEngine.java
@@ -24,6 +24,7 @@
 import org.opensearch.index.seqno.SeqNoStats;
 import org.opensearch.index.seqno.SequenceNumbers;
 import org.opensearch.index.translog.Translog;
+import org.opensearch.index.translog.TranslogCorruptedException;
 import org.opensearch.index.translog.TranslogDeletionPolicy;
 import org.opensearch.index.translog.TranslogException;
 import org.opensearch.index.translog.TranslogManager;
@@ -73,6 +74,7 @@ public NRTReplicationEngine(EngineConfig engineConfig) {
         store.incRef();
         NRTReplicationReaderManager readerManager = null;
         WriteOnlyTranslogManager translogManagerRef = null;
+        boolean success = false;
         try {
             this.replicaFileTracker = new ReplicaFileTracker(store::deleteQuiet);
             this.lastCommittedSegmentInfos = store.readLastCommittedSegmentsInfo();
@@ -126,9 +128,17 @@ public void onAfterTranslogSync() {
                 engineConfig.getPrimaryModeSupplier()
             );
             this.translogManager = translogManagerRef;
-        } catch (IOException e) {
-            IOUtils.closeWhileHandlingException(store::decRef, readerManager, translogManagerRef);
+            success = true;
+        } catch (IOException | TranslogCorruptedException e) {
             throw new EngineCreationFailureException(shardId, "failed to create engine", e);
+        } finally {
+            if (success == false) {
+                IOUtils.closeWhileHandlingException(readerManager, translogManagerRef);
+                if (isClosed.get() == false) {
+                    // failure, we need to dec the store reference
+                    store.decRef();
+                }
+            }
         }
     }
 

diff --git a/server/src/test/java/org/opensearch/index/engine/NRTReplicationEngineTests.java b/server/src/test/java/org/opensearch/index/engine/NRTReplicationEngineTests.java
@@ -75,6 +75,28 @@ public void testCreateEngine() throws IOException {
         }
     }
 
+    public void testCreateEngineWithException() throws IOException {
+        final AtomicLong globalCheckpoint = new AtomicLong(SequenceNumbers.NO_OPS_PERFORMED);
+        final Store nrtEngineStore = createStore(INDEX_SETTINGS, newDirectory());
+        try {
+            // Passing null translogPath to induce failure
+            final EngineConfig replicaConfig = config(
+                defaultSettings,
+                nrtEngineStore,
+                null,
+                NoMergePolicy.INSTANCE,
+                null,
+                null,
+                globalCheckpoint::get
+            );
+            new NRTReplicationEngine(replicaConfig);
+        } catch (Exception e) {
+            // Ignore as engine creation will fail
+        }
+        assertEquals(1, nrtEngineStore.refCount());
+        nrtEngineStore.close();
+    }
+
     public void testEngineWritesOpsToTranslog() throws Exception {
         final AtomicLong globalCheckpoint = new AtomicLong(SequenceNumbers.NO_OPS_PERFORMED);