Merge pull request elastic#15788 from s1monw/dont_delete_tlog_file

Never delete translog-N.tlog file when creation fails

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

@@ -163,6 +163,21 @@ public Translog(TranslogConfig config) throws IOException {
         try {
             if (translogGeneration != null) {
                 final Checkpoint checkpoint = readCheckpoint();
+                final Path nextTranslogFile = location.resolve(getFilename(checkpoint.generation + 1));
+                final Path currentCheckpointFile = location.resolve(getCommitCheckpointFileName(checkpoint.generation));
+                // this is special handling for error condition when we create a new writer but we fail to bake
+                // the newly written file (generation+1) into the checkpoint. This is still a valid state
+                // we just need to cleanup before we continue
+                // we hit this before and then blindly deleted the new generation even though we managed to bake it in and then hit this:
+                // https://discuss.elastic.co/t/cannot-recover-index-because-of-missing-tanslog-files/38336 as an example
+                //
+                // For this to happen we must have already copied the translog.ckp file into translog-gen.ckp so we first check if that file exists
+                // if not we don't even try to clean it up and wait until we fail creating it
+                assert Files.exists(nextTranslogFile) == false || Files.size(nextTranslogFile) <= TranslogWriter.getHeaderLength(translogUUID) : "unexpected translog file: [" + nextTranslogFile + "]";
+                if (Files.exists(currentCheckpointFile) // current checkpoint is already copied
+                    && Files.deleteIfExists(nextTranslogFile)) { // delete it and log a warning
+                    logger.warn("deleted previously created, but not yet committed, next generation [{}]. This can happen due to a tragic exception when creating a new generation", nextTranslogFile.getFileName());
+                }
                 this.recoveredTranslogs = recoverFromFiles(translogGeneration, checkpoint);
                 if (recoveredTranslogs.isEmpty()) {
                     throw new IllegalStateException("at least one reader must be recovered");

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

@@ -69,9 +69,17 @@ public TranslogWriter(ShardId shardId, long generation, ChannelReference channel
         this.lastSyncedOffset = channelReference.getChannel().position();;
     }
 
+    static int getHeaderLength(String translogUUID) {
+        return getHeaderLength(new BytesRef(translogUUID).length);
+    }
+
+    private static int getHeaderLength(int uuidLength) {
+        return CodecUtil.headerLength(TRANSLOG_CODEC) + uuidLength  + RamUsageEstimator.NUM_BYTES_INT;
+    }
+
     public static TranslogWriter create(Type type, ShardId shardId, String translogUUID, long fileGeneration, Path file, Callback<ChannelReference> onClose, int bufferSize, ChannelFactory channelFactory) throws IOException {
         final BytesRef ref = new BytesRef(translogUUID);
-        final int headerLength = CodecUtil.headerLength(TRANSLOG_CODEC) + ref.length + RamUsageEstimator.NUM_BYTES_INT;
+        final int headerLength = getHeaderLength(ref.length);
         final FileChannel channel = channelFactory.open(file);
         try {
             // This OutputStreamDataOutput is intentionally not closed because
@@ -80,17 +88,14 @@ public static TranslogWriter create(Type type, ShardId shardId, String translogU
             CodecUtil.writeHeader(out, TRANSLOG_CODEC, VERSION);
             out.writeInt(ref.length);
             out.writeBytes(ref.bytes, ref.offset, ref.length);
-            channel.force(false);
+            channel.force(true);
             writeCheckpoint(headerLength, 0, file.getParent(), fileGeneration, StandardOpenOption.WRITE);
             final TranslogWriter writer = type.create(shardId, fileGeneration, new ChannelReference(file, fileGeneration, channel, onClose), bufferSize);
             return writer;
         } catch (Throwable throwable){
+            // if we fail to bake the file-generation into the checkpoint we stick with the file and once we recover and that
+            // file exists we remove it. We only apply this logic to the checkpoint.generation+1 any other file with a higher generation is an error condition
             IOUtils.closeWhileHandlingException(channel);
-            try {
-                Files.delete(file); // remove the file as well
-            } catch (IOException ex) {
-                throwable.addSuppressed(ex);
-            }
             throw throwable;
         }
     }

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

@@ -1790,4 +1790,96 @@ protected TranslogWriter createWriter(long fileGeneration) throws IOException {
             // all is well
         }
     }
+
+    public void testRecoverWithUnbackedNextGen() throws IOException {
+        translog.add(new Translog.Index("test", "" + 0, Integer.toString(0).getBytes(Charset.forName("UTF-8"))));
+        Translog.TranslogGeneration translogGeneration = translog.getGeneration();
+        translog.close();
+        TranslogConfig config = translog.getConfig();
+
+        Path ckp = config.getTranslogPath().resolve(Translog.CHECKPOINT_FILE_NAME);
+        Checkpoint read = Checkpoint.read(ckp);
+        Files.copy(ckp, config.getTranslogPath().resolve(Translog.getCommitCheckpointFileName(read.generation)));
+        Files.createFile(config.getTranslogPath().resolve("translog-" + (read.generation + 1) + ".tlog"));
+        config.setTranslogGeneration(translogGeneration);
+        try (Translog tlog = new Translog(config)) {
+            assertNotNull(translogGeneration);
+            assertFalse(tlog.syncNeeded());
+            try (Translog.Snapshot snapshot = tlog.newSnapshot()) {
+                for (int i = 0; i < 1; i++) {
+                    Translog.Operation next = snapshot.next();
+                    assertNotNull("operation " + i + " must be non-null", next);
+                    assertEquals("payload missmatch", i, Integer.parseInt(next.getSource().source.toUtf8()));
+                }
+            }
+            tlog.add(new Translog.Index("test", "" + 1, Integer.toString(1).getBytes(Charset.forName("UTF-8"))));
+        }
+        try (Translog tlog = new Translog(config)) {
+            assertNotNull(translogGeneration);
+            assertFalse(tlog.syncNeeded());
+            try (Translog.Snapshot snapshot = tlog.newSnapshot()) {
+                for (int i = 0; i < 2; i++) {
+                    Translog.Operation next = snapshot.next();
+                    assertNotNull("operation " + i + " must be non-null", next);
+                    assertEquals("payload missmatch", i, Integer.parseInt(next.getSource().source.toUtf8()));
+                }
+            }
+        }
+    }
+
+    public void testRecoverWithUnbackedNextGenInIllegalState() throws IOException {
+        translog.add(new Translog.Index("test", "" + 0, Integer.toString(0).getBytes(Charset.forName("UTF-8"))));
+        Translog.TranslogGeneration translogGeneration = translog.getGeneration();
+        translog.close();
+        TranslogConfig config = translog.getConfig();
+        Path ckp = config.getTranslogPath().resolve(Translog.CHECKPOINT_FILE_NAME);
+        Checkpoint read = Checkpoint.read(ckp);
+        // don't copy the new file
+        Files.createFile(config.getTranslogPath().resolve("translog-" + (read.generation + 1) + ".tlog"));
+        config.setTranslogGeneration(translogGeneration);
+
+        try  {
+            Translog tlog = new Translog(config);
+            fail("file already exists?");
+        } catch (TranslogException ex) {
+            // all is well
+            assertEquals(ex.getMessage(), "failed to create new translog file");
+            assertEquals(ex.getCause().getClass(), FileAlreadyExistsException.class);
+        }
+    }
+    public void testRecoverWithUnbackedNextGenAndFutureFile() throws IOException {
+        translog.add(new Translog.Index("test", "" + 0, Integer.toString(0).getBytes(Charset.forName("UTF-8"))));
+        Translog.TranslogGeneration translogGeneration = translog.getGeneration();
+        translog.close();
+        TranslogConfig config = translog.getConfig();
+
+        Path ckp = config.getTranslogPath().resolve(Translog.CHECKPOINT_FILE_NAME);
+        Checkpoint read = Checkpoint.read(ckp);
+        Files.copy(ckp, config.getTranslogPath().resolve(Translog.getCommitCheckpointFileName(read.generation)));
+        Files.createFile(config.getTranslogPath().resolve("translog-" + (read.generation + 1) + ".tlog"));
+        // we add N+1 and N+2 to ensure we only delete the N+1 file and never jump ahead and wipe without the right condition
+        Files.createFile(config.getTranslogPath().resolve("translog-" + (read.generation + 2) + ".tlog"));
+        config.setTranslogGeneration(translogGeneration);
+        try (Translog tlog = new Translog(config)) {
+            assertNotNull(translogGeneration);
+            assertFalse(tlog.syncNeeded());
+            try (Translog.Snapshot snapshot = tlog.newSnapshot()) {
+                for (int i = 0; i < 1; i++) {
+                    Translog.Operation next = snapshot.next();
+                    assertNotNull("operation " + i + " must be non-null", next);
+                    assertEquals("payload missmatch", i, Integer.parseInt(next.getSource().source.toUtf8()));
+                }
+            }
+            tlog.add(new Translog.Index("test", "" + 1, Integer.toString(1).getBytes(Charset.forName("UTF-8"))));
+        }
+
+        try  {
+            Translog tlog = new Translog(config);
+            fail("file already exists?");
+        } catch (TranslogException ex) {
+            // all is well
+            assertEquals(ex.getMessage(), "failed to create new translog file");
+            assertEquals(ex.getCause().getClass(), FileAlreadyExistsException.class);
+        }
+    }
 }