[CORE] Handle truncated translog gracefully

We used to handle truncated translogs in a better manner (assuming that
the node was killed halfway through writing an operation and discarding
the last operation). This brings back that behavior by catching an
`EOFException` during the stream reading and throwing a
`TruncatedTranslogException` which can be safely ignored in
`IndexShardGateway`.

Fixes elastic#9699

src/main/java/org/elasticsearch/index/gateway/IndexShardGateway.java

@@ -39,5 +39,4 @@ public interface IndexShardGateway extends IndexShardComponent, CloseableIndexCo
      * Recovers the state of the shard from the gateway.
      */
     void recover(boolean indexShouldExists, RecoveryState recoveryState) throws IndexShardGatewayRecoveryException;
-
 }

src/main/java/org/elasticsearch/index/gateway/local/LocalIndexShardGateway.java

@@ -255,7 +255,7 @@ public void recover(boolean indexShouldExists, RecoveryState recoveryState) thro
                             in.readInt(); // ignored opSize
                         }
                         operation = stream.read(in);
-                    } catch (EOFException e) {
+                    } catch (TruncatedTranslogException|EOFException e) {
                         // ignore, not properly written the last op
                         logger.trace("ignoring translog EOF exception, the last operation was not properly written ([{}])", e.getMessage());
                         break;

src/main/java/org/elasticsearch/index/translog/ChecksummedTranslogStream.java

@@ -66,6 +66,8 @@ public Translog.Operation read(StreamInput inStream) throws IOException {
             Translog.Operation.Type type = Translog.Operation.Type.fromId(in.readByte());
             operation = TranslogStreams.newOperationFromType(type);
             operation.readFrom(in);
+        } catch (EOFException e) {
+            throw new TruncatedTranslogException("reached premature end of file, translog is truncated", e);
         } catch (AssertionError|Exception e) {
             throw new TranslogCorruptedException("translog corruption while reading from stream", e);
         }

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

@@ -35,6 +35,7 @@
 import org.junit.Test;
 
 import java.io.File;
+import java.io.EOFException;
 import java.io.IOException;
 import java.io.RandomAccessFile;
 import java.nio.file.DirectoryStream;
@@ -438,6 +439,54 @@ public void testTranslogChecksums() throws Exception {
         assertThat("at least one corruption was caused and caught", corruptionsCaught.get(), greaterThanOrEqualTo(1));
     }
 
+    @Test
+    public void testTruncatedTranslogs() throws Exception {
+        List<Translog.Location> locations = newArrayList();
+
+        int translogOperations = randomIntBetween(10, 100);
+        for (int op = 0; op < translogOperations; op++) {
+            String ascii = randomAsciiOfLengthBetween(1, 50);
+            locations.add(translog.add(new Translog.Create("test", "" + op, ascii.getBytes("UTF-8"))));
+        }
+        translog.sync();
+
+        truncateTranslogs(translogFileDirectory());
+
+        AtomicInteger truncations = new AtomicInteger(0);
+        for (Translog.Location location : locations) {
+            try {
+                translog.read(location);
+            } catch (ElasticsearchException e) {
+                if (e.getCause() instanceof EOFException) {
+                    truncations.incrementAndGet();
+                } else {
+                    throw e;
+                }
+            }
+        }
+        assertThat("at least one truncation was caused and caught", truncations.get(), greaterThanOrEqualTo(1));
+    }
+
+    /**
+     * Randomly truncate some bytes in the translog files
+     */
+    private void truncateTranslogs(String directory) throws Exception {
+        File[] files = new File(directory).listFiles();
+        if (files != null) {
+            for (File file : files) {
+                if (file.getName().startsWith("translog-")) {
+                    RandomAccessFile f = new RandomAccessFile(file, "rw");
+                    long prevSize = f.length();
+                    long newSize = prevSize - randomIntBetween(1, (int) prevSize / 2);
+                    logger.info("--> truncating {}, prev: {}, now: {}", file, prevSize, newSize);
+                    f.setLength(newSize);
+                    f.close();
+                }
+            }
+        }
+    }
+
+
     /**
      * Randomly overwrite some bytes in the translog files
      */

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

@@ -26,6 +26,8 @@
 
 import java.io.EOFException;
 import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Path;
 
 import static org.hamcrest.Matchers.equalTo;
 
@@ -144,4 +146,26 @@ public void testCorruptedTranslogs() throws Exception {
         }
 
     }
+
+    @Test
+    public void testTruncatedTranslog() throws Exception {
+        try {
+            File translogFile = getResource("/org/elasticsearch/index/translog/translog-v1-truncated.binary");
+            assertThat("test file should exist", Files.exists(translogFile.toPath()), equalTo(true));
+            TranslogStream stream = TranslogStreams.translogStreamFor(translogFile);
+            try (StreamInput in = stream.openInput(translogFile)) {
+                while (true) {
+                    try {
+                        stream.read(in);
+                    } catch (EOFException e) {
+                        break;
+                    }
+                }
+            }
+            fail("should have thrown an exception about the body being truncated");
+        } catch (TruncatedTranslogException e) {
+            assertThat("translog truncated: " + e.getMessage(),
+                    e.getMessage().contains("reached premature end of file, translog is truncated"), equalTo(true));
+        }
+    }
 }