Allow to trim all ops above a certain seq# with a term lower than X - fixes on PR elastic#3

Relates to elastic#10708

Author: Vladimir Dolzhenko

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

@@ -706,27 +706,16 @@ public void trim(long belowTerm, long aboveSeqNo) throws IOException {
 
             // update all existed ones (if it is necessary) as checkpoint and reader are immutable
             final List<TranslogReader> newReaders = new ArrayList<>(readers.size());
-            for (TranslogReader reader : readers) {
-                final Checkpoint checkpoint = reader.getCheckpoint();
-                final Path checkpointFile = location.resolve(getCommitCheckpointFileName(checkpoint.generation));
-
-                final TranslogReader newReader;
-                if (reader.getPrimaryTerm() < belowTerm
-                    && checkpoint.maxSeqNo != SequenceNumbers.NO_OPS_PERFORMED
-                    && (aboveSeqNo < checkpoint.trimmedAboveSeqNo || checkpoint.trimmedAboveSeqNo == SequenceNumbers.UNASSIGNED_SEQ_NO)) {
-                    final Checkpoint newCheckpoint = new Checkpoint(checkpoint.offset, checkpoint.numOps,
-                        checkpoint.generation, checkpoint.minSeqNo, checkpoint.maxSeqNo,
-                        checkpoint.globalCheckpoint, checkpoint.minTranslogGeneration, aboveSeqNo);
-                    Checkpoint.write(FileChannel::open, checkpointFile, newCheckpoint, StandardOpenOption.WRITE);
-
-                    IOUtils.fsync(checkpointFile, false);
-                    IOUtils.fsync(checkpointFile.getParent(), true);
-
-                    newReader = reader.withNewCheckpoint(newCheckpoint);
-                } else {
-                    newReader = reader;
+            try {
+                for (TranslogReader reader : readers) {
+                    final TranslogReader newReader = reader.closeIntoTrimmedReader(belowTerm, aboveSeqNo, getChannelFactory());
+                    newReaders.add(newReader);
                 }
-                newReaders.add(newReader);
+            } catch (IOException e){
+                IOUtils.closeWhileHandlingException(newReaders);
+                IOUtils.closeWhileHandlingException(current);
+                IOUtils.closeWhileHandlingException(readers);
+                throw e;
             }
 
             this.readers.clear();

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

@@ -21,23 +21,28 @@
 
 import org.apache.lucene.store.AlreadyClosedException;
 import org.elasticsearch.common.io.Channels;
+import org.elasticsearch.core.internal.io.IOUtils;
+import org.elasticsearch.index.seqno.SequenceNumbers;
 
 import java.io.Closeable;
 import java.io.EOFException;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import static org.elasticsearch.index.translog.Translog.getCommitCheckpointFileName;
+
 /**
  * an immutable translog filereader
  */
 public class TranslogReader extends BaseTranslogReader implements Closeable {
     protected final long length;
     private final int totalOperations;
     private final Checkpoint checkpoint;
-    protected final AtomicBoolean closed;
+    protected final AtomicBoolean closed = new AtomicBoolean(false);
 
     /**
      * Create a translog writer against the specified translog file channel.
@@ -48,24 +53,10 @@ public class TranslogReader extends BaseTranslogReader implements Closeable {
      * @param header     the header of the translog file
      */
     TranslogReader(final Checkpoint checkpoint, final FileChannel channel, final Path path, final TranslogHeader header) {
-        this(checkpoint, channel, path, header,  new AtomicBoolean(false));
-    }
-
-    /**
-     * Create a translog writer against the specified translog file channel.
-     *
-     * @param checkpoint the translog checkpoint
-     * @param channel    the translog file channel to open a translog reader against
-     * @param path       the path to the translog
-     * @param header     the header of the translog file
-     *
-     */
-    TranslogReader(final Checkpoint checkpoint, final FileChannel channel, final Path path, final TranslogHeader header, AtomicBoolean closed) {
         super(checkpoint.generation, channel, path, header);
         this.length = checkpoint.offset;
         this.totalOperations = checkpoint.numOps;
         this.checkpoint = checkpoint;
-        this.closed = closed;
     }
 
     /**
@@ -85,10 +76,27 @@ public static TranslogReader open(
     }
 
     /**
-     * Create a new reader with new checkoint that shares resources with current one
+     * Closes current reader and creates new one with new checkoint and same file channel
      */
-    TranslogReader withNewCheckpoint(final Checkpoint newCheckpoint){
-        return new TranslogReader(newCheckpoint, channel, path, header,  closed);
+    TranslogReader closeIntoTrimmedReader(long belowTerm, long aboveSeqNo, ChannelFactory channelFactory) throws IOException {
+        if (!closed.get()
+            && getPrimaryTerm() < belowTerm
+            && checkpoint.maxSeqNo != SequenceNumbers.NO_OPS_PERFORMED
+            && (aboveSeqNo < checkpoint.trimmedAboveSeqNo || checkpoint.trimmedAboveSeqNo == SequenceNumbers.UNASSIGNED_SEQ_NO)) {
+            final Path checkpointFile = path.getParent().resolve(getCommitCheckpointFileName(checkpoint.generation));
+            final Checkpoint newCheckpoint = new Checkpoint(checkpoint.offset, checkpoint.numOps,
+                checkpoint.generation, checkpoint.minSeqNo, checkpoint.maxSeqNo,
+                checkpoint.globalCheckpoint, checkpoint.minTranslogGeneration, aboveSeqNo);
+            Checkpoint.write(channelFactory, checkpointFile, newCheckpoint, StandardOpenOption.WRITE);
+
+            IOUtils.fsync(checkpointFile, false);
+            IOUtils.fsync(checkpointFile.getParent(), true);
+
+            closed.set(true);
+
+            return new TranslogReader(newCheckpoint, channel, path, header);
+        }
+        return this;
     }
 
     public long sizeInBytes() {

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

@@ -1545,6 +1545,49 @@ public void testTrimmedSnapshot() throws Exception {
         assertThat(expectedSeqNo, hasSize(0));
     }
 
+    public void testExceptionOnTrimAboveSeqNo( ) throws Exception {
+        Path tempDir = createTempDir();
+        final FailSwitch fail = new FailSwitch();
+        fail.failNever();
+        TranslogConfig config = getTranslogConfig(tempDir);
+        final Translog failableTLog = getFailableTranslog(fail, config, randomBoolean(), false, null, createTranslogDeletionPolicy());
+
+        List<Translog.Index> operations = new ArrayList<>();
+        int translogOperations = randomIntBetween(10, 100);
+        int maxTrimmedSeqNo = translogOperations - randomIntBetween(4, 8);
+
+        for (int op = 0; op < translogOperations; op++) {
+            String ascii = randomAlphaOfLengthBetween(1, 50);
+            Translog.Index operation = new Translog.Index("test", "" + op, op,
+                primaryTerm.get(), ascii.getBytes("UTF-8"));
+            operations.add(operation);
+        }
+        // shuffle a bit - move several first items to the end
+        for(int i = 0, len = randomIntBetween(5, 10); i < len; i++){
+            operations.add(operations.remove(0));
+        }
+
+        for (Translog.Index operation : operations) {
+            failableTLog.add(operation);
+        }
+
+        failableTLog.rollGeneration();
+        fail.failAlways();
+        try {
+            failableTLog.trim(primaryTerm.get() + 1, maxTrimmedSeqNo);
+            fail();
+        } catch (MockDirectoryWrapper.FakeIOException ex) {
+            // all is fine
+        }
+
+        try {
+            failableTLog.newSnapshot();
+            fail();
+        } catch (AlreadyClosedException e){
+            assertThat(e.getMessage(), is("translog [" + failableTLog.currentFileGeneration() + "] is already closed"));
+        }
+    }
+
     public void testLocationHashCodeEquals() throws IOException {
         List<Translog.Location> locations = new ArrayList<>();
         List<Translog.Location> locations2 = new ArrayList<>();