Allow double-closing of FSTranslog

the translog might be reused across engines which is currently a problem
in the design such that we have to allow calls to `close` more than once.
This moves the closed check for snapshot on the actual file to exit the loop.

Relates to #10807

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

@@ -238,6 +238,11 @@ public void updateBufferSize(int bufferSize) {
         }
     }
 
+    @Override
+    public boolean closed() {
+        return this.closed.get();
+    }
+
     class WrapperOutputStream extends OutputStream {
 
         @Override

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

@@ -45,7 +45,6 @@
 import java.nio.channels.ClosedChannelException;
 import java.nio.file.Path;
 import java.nio.file.*;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
@@ -85,8 +84,6 @@ 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 {
@@ -147,23 +144,21 @@ public void updateBuffer(ByteSizeValue bufferSize) {
     }
 
     private void close(boolean delete) {
-        if (closed.compareAndSet(false, true)) {
-            if (indexSettingsService != null) {
-                indexSettingsService.removeListener(applySettings);
+        if (indexSettingsService != null) {
+            indexSettingsService.removeListener(applySettings);
+        }
+        rwl.writeLock().lock();
+        try {
+            FsTranslogFile current1 = this.current;
+            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();
+            current1 = this.trans;
+            if (current1 != null) {
+                current1.close(delete);
             }
+        } finally {
+            rwl.writeLock().unlock();
         }
     }
 
@@ -405,20 +400,22 @@ public Location add(Operation operation) throws TranslogException {
             rwl.readLock().unlock();
             if (!released && out != null) {
                 Releasables.close(out.bytes());
-            }
+
         }
     }
 
     @Override
     public FsChannelSnapshot snapshot() throws TranslogException {
         while (true) {
-            if (closed.get()) {
-                throw new TranslogException(shardId, "translog is already closed");
-            }
+            FsTranslogFile current = this.current;
             FsChannelSnapshot snapshot = current.snapshot();
             if (snapshot != null) {
                 return snapshot;
             }
+            if (current.closed() && this.current == current) {
+                // check if we are closed and if we are still current - then this translog is closed and we can exit
+                throw new TranslogException(shardId, "current translog is already closed");
+            }
             Thread.yield();
         }
     }

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

@@ -80,4 +80,6 @@ public static Type fromString(String type) throws ElasticsearchIllegalArgumentEx
     boolean syncNeeded();
 
     TranslogStream getStream();
+
+    public boolean closed();
 }

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

@@ -182,4 +182,9 @@ public String toString() {
                 ", operationCounter=" + operationCounter +
                 '}';
     }
+
+    public boolean closed() {
+        return this.closed.get();
+    }
+
 }

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

@@ -318,7 +318,7 @@ public void testSnapshotOnClosedTranslog() throws IOException {
             Translog.Snapshot snapshot = translog.snapshot();
             fail("translog is closed");
         } catch (TranslogException ex) {
-            assertEquals(ex.getMessage(), "[index][1] translog is already closed");
+            assertEquals(ex.getMessage(), "[index][1] current translog is already closed");
         }
     }