Translog can delete valid .ckp file when views are closed after the t…

…ranslog (#19035)

There is simply a coding bug that only happens if translog views are closed
after the translog itself is closed. this can happen for instance if we hit
a disk full exception and try to repeatedly recover the translog. This will
cause a translog-N.ckp file to be deleted since the wrong generation is used
to generate the path to delete. This seems like a copy/past problem.

This bug doesn't affect 5.0

Relates to #16495

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

@@ -731,7 +731,7 @@ public void handle(ChannelReference channelReference) {
                         if (isReferencedGeneration(generation) == false) {
                             logger.trace("delete translog file - not referenced and not current anymore {}", path);
                             IOUtils.deleteFilesIgnoringExceptions(path);
-                            IOUtils.deleteFilesIgnoringExceptions(path.resolveSibling(getCommitCheckpointFileName(channelReference.getGeneration())));
+                            IOUtils.deleteFilesIgnoringExceptions(path.resolveSibling(getCommitCheckpointFileName(generation)));
                         }
                     }
                 }

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

@@ -2059,4 +2059,21 @@ public void testPullViewOnClosed() throws IOException {
             // all is well
         }
     }
+
+    public void testPendingDelete() throws IOException {
+        translog.add(new Translog.Create("test", "1", new byte[]{1}));
+        translog.prepareCommit();
+        Translog.TranslogGeneration generation = translog.getGeneration();
+        TranslogConfig config = translog.getConfig();
+        translog.close();
+        config.setTranslogGeneration(generation);
+        translog = new Translog(config);
+        translog.add(new Translog.Create("test", "2", new byte[]{2}));
+        translog.prepareCommit();
+        Translog.View view = translog.newView();
+        translog.add(new Translog.Create("test", "3", new byte[]{3}));
+        translog.close();
+        IOUtils.close(view);
+        translog = new Translog(config);
+    }
 }