Merge pull request #15762 from s1monw/issues/15754

Close recovered translog readers if createWriter fails

core/src/main/java/org/elasticsearch/index/translog/Translog.java

@@ -167,8 +167,19 @@ public class Translog extends AbstractIndexShardComponent implements IndexShardC
                 if (recoveredTranslogs.isEmpty()) {
                     throw new IllegalStateException("at least one reader must be recovered");
                 }
+                boolean success = false;
+                try {
                     current = createWriter(checkpoint.generation + 1);
                     this.lastCommittedTranslogFileGeneration = translogGeneration.translogFileGeneration;
+                    success = true;
+                } finally {
+                    // we have to close all the recovered ones otherwise we leak file handles here
+                    // for instance if we have a lot of tlog and we can't create the writer we keep on holding
+                    // on to all the uncommitted tlog files if we don't close
+                    if (success == false) {
+                        IOUtils.closeWhileHandlingException(recoveredTranslogs);
+                    }
+                }
             } else {
                 this.recoveredTranslogs = Collections.emptyList();
                 IOUtils.rm(location);

core/src/main/java/org/elasticsearch/index/translog/TranslogReader.java

@@ -138,7 +138,7 @@ public abstract class TranslogReader implements Closeable, Comparable<TranslogRe
     abstract protected void readBytes(ByteBuffer buffer, long position) throws IOException;
 
     @Override
-    public void close() throws IOException {
+    public final void close() throws IOException {
         if (closed.compareAndSet(false, true)) {
             channelReference.decRef();
         }

core/src/test/java/org/elasticsearch/index/translog/TranslogTests.java

@@ -1590,4 +1590,27 @@ public class TranslogTests extends ESTestCase {
     private static final class UnknownException extends RuntimeException {
 
     }
+
+    // see https://github.com/elastic/elasticsearch/issues/15754
+    public void testFailWhileCreateWriteWithRecoveredTLogs() throws IOException {
+        Path tempDir = createTempDir();
+        TranslogConfig config = getTranslogConfig(tempDir);
+        Translog translog = new Translog(config);
+        translog.add(new Translog.Index("test", "boom", "boom".getBytes(Charset.forName("UTF-8"))));
+        Translog.TranslogGeneration generation = translog.getGeneration();
+        translog.close();
+        config.setTranslogGeneration(generation);
+        try {
+            new Translog(config) {
+                @Override
+                protected TranslogWriter createWriter(long fileGeneration) throws IOException {
+                    throw new MockDirectoryWrapper.FakeIOException();
+                }
+            };
+            // if we have a LeakFS here we fail if not all resources are closed
+            fail("should have been failed");
+        } catch (MockDirectoryWrapper.FakeIOException ex) {
+            // all is well
+        }
+    }
 }