[TRANSLOG] Fail #snapshot if translog is closed

If the translog is closed while a snapshot opertion is in progress
we must fail the snapshot operation otherwise we end up in an endless
loop.

Closes #10807

src/main/java/org/elasticsearch/index/translog/fs/FsTranslog.java

@@ -50,6 +50,7 @@ import java.nio.file.*;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.regex.Matcher;
@@ -93,7 +94,7 @@ public class FsTranslog extends AbstractIndexShardComponent implements Translog
 
     private final ApplySettings applySettings = new ApplySettings();
 
-
+    private final AtomicBoolean closed = new AtomicBoolean(false);
 
     @Inject
     public FsTranslog(ShardId shardId, @IndexSettings Settings indexSettings, IndexSettingsService indexSettingsService,
@@ -140,14 +141,16 @@ public class FsTranslog extends AbstractIndexShardComponent implements Translog
 
     @Override
     public void close() throws IOException {
-        if (indexSettingsService != null) {
-            indexSettingsService.removeListener(applySettings);
-        }
-        rwl.writeLock().lock();
-        try {
-            IOUtils.close(this.trans, this.current);
-        } finally {
-            rwl.writeLock().unlock();
+        if (closed.compareAndSet(false, true)) {
+            if (indexSettingsService != null) {
+                indexSettingsService.removeListener(applySettings);
+            }
+            rwl.writeLock().lock();
+            try {
+                IOUtils.close(this.trans, this.current);
+            } finally {
+                rwl.writeLock().unlock();
+            }
         }
     }
 
@@ -355,6 +358,9 @@ public class FsTranslog extends AbstractIndexShardComponent implements Translog
     @Override
     public FsChannelSnapshot snapshot() throws TranslogException {
         while (true) {
+            if (closed.get()) {
+                throw new TranslogException(shardId, "translog is already closed");
+            }
             FsChannelSnapshot snapshot = current.snapshot();
             if (snapshot != null) {
                 return snapshot;

src/test/java/org/elasticsearch/index/translog/AbstractSimpleTranslogTests.java

@@ -332,6 +332,17 @@ public abstract class AbstractSimpleTranslogTests extends ElasticsearchTestCase
         snapshot.close();
     }
 
+    public void testSnapshotOnClosedTranslog() throws IOException {
+        assertTrue(Files.exists(translogDir.resolve("translog-1")));
+        translog.add(new Translog.Create("test", "1", new byte[]{1}));
+        translog.close();
+        try {
+            Translog.Snapshot snapshot = translog.snapshot();
+        } catch (TranslogException ex) {
+            assertEquals(ex.getMessage(), "translog is already closed");
+        }
+    }
+
     @Test
     public void deleteOnRollover() throws IOException {
         translog.add(new Translog.Create("test", "1", new byte[]{1}));