From f6020a6ff29824757855348157933e5cd90b49ba Mon Sep 17 00:00:00 2001
From: Simon Willnauer <simonw@apache.org>
Date: Mon, 4 Jan 2016 23:40:20 +0100
Subject: [PATCH] Close recovered translog readers if createWriter fails

If we fail to create a writer all recovered translog readers are not
closed today which causes all open files to leak.

Closes #15754
---
 .../index/translog/Translog.java              | 15 ++++++++++--
 .../index/translog/TranslogReader.java        |  2 +-
 .../index/translog/TranslogTests.java         | 23 +++++++++++++++++++
 3 files changed, 37 insertions(+), 3 deletions(-)

diff --git a/core/src/main/java/org/elasticsearch/index/translog/Translog.java b/core/src/main/java/org/elasticsearch/index/translog/Translog.java
index c2cb3c19af8..3f8f0ab54df 100644
--- a/core/src/main/java/org/elasticsearch/index/translog/Translog.java
+++ b/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");
                 }
-                current = createWriter(checkpoint.generation + 1);
-                this.lastCommittedTranslogFileGeneration = translogGeneration.translogFileGeneration;
+                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);
diff --git a/core/src/main/java/org/elasticsearch/index/translog/TranslogReader.java b/core/src/main/java/org/elasticsearch/index/translog/TranslogReader.java
index d7077fd90ad..71dff6ec36e 100644
--- a/core/src/main/java/org/elasticsearch/index/translog/TranslogReader.java
+++ b/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();
         }
diff --git a/core/src/test/java/org/elasticsearch/index/translog/TranslogTests.java b/core/src/test/java/org/elasticsearch/index/translog/TranslogTests.java
index c1d2b2e1997..1da2b7bf3c8 100644
--- a/core/src/test/java/org/elasticsearch/index/translog/TranslogTests.java
+++ b/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
+        }
+    }
 }