[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 elastic#10807

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

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

@@ -39,14 +39,14 @@
 import org.elasticsearch.index.translog.TranslogException;
 import org.elasticsearch.index.translog.TranslogStats;
 import org.elasticsearch.index.translog.TranslogStreams;
-import org.elasticsearch.index.translog.*;
 
 import java.io.File;
 import java.io.IOException;
 import java.nio.channels.ClosedChannelException;
 import java.nio.file.Path;
 import java.util.concurrent.ThreadLocalRandom;
 import java.nio.file.*;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
@@ -86,6 +86,8 @@ public void onRefreshSettings(Settings settings) {
 
     private final ApplySettings applySettings = new ApplySettings();
 
+    private final AtomicBoolean closed = new AtomicBoolean(false);
+
     @Inject
     public FsTranslog(ShardId shardId, @IndexSettings Settings indexSettings, IndexSettingsService indexSettingsService,
                       BigArrays bigArrays, IndexStore indexStore) throws IOException {
@@ -146,21 +148,23 @@ public void updateBuffer(ByteSizeValue bufferSize) {
     }
 
     private void close(boolean delete) {
-        if (indexSettingsService != null) {
-            indexSettingsService.removeListener(applySettings);
-        }
-        rwl.writeLock().lock();
-        try {
-            FsTranslogFile current1 = this.current;
-            if (current1 != null) {
-                current1.close(delete);
+        if (closed.compareAndSet(false, true)) {
+            if (indexSettingsService != null) {
+                indexSettingsService.removeListener(applySettings);
             }
-            current1 = this.trans;
-            if (current1 != null) {
-                current1.close(delete);
+            rwl.writeLock().lock();
+            try {
+                FsTranslogFile current1 = this.current;
+                if (current1 != null) {
+                    current1.close(delete);
+                }
+                current1 = this.trans;
+                if (current1 != null) {
+                    current1.close(delete);
+                }
+            } finally {
+                rwl.writeLock().unlock();
             }
-        } finally {
-            rwl.writeLock().unlock();
         }
     }
 
@@ -413,6 +417,9 @@ public Location add(Operation operation) throws TranslogException {
     @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

@@ -311,6 +311,17 @@ public void testSnapshotWithNewTranslog() {
         snapshot.close();
     }
 
+    public void testSnapshotOnClosedTranslog() throws IOException {
+        translog.add(new Translog.Create("test", "1", new byte[]{1}));
+        translog.close();
+        try {
+            Translog.Snapshot snapshot = translog.snapshot();
+            fail("translog is closed");
+        } catch (TranslogException ex) {
+            assertEquals(ex.getMessage(), "[index][1] translog is already closed");
+        }
+    }
+
     @Test
     public void testSnapshotWithSeekTo() {
         Translog.Snapshot snapshot = translog.snapshot();