Allow to trim all ops above a certain seq# with a term lower than X

Relates to #10708

Author: vladimirdolzhenko

server/src/main/java/org/elasticsearch/action/resync/ResyncReplicationRequest.java

@@ -22,6 +22,7 @@
 import org.elasticsearch.action.support.replication.ReplicatedWriteRequest;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.translog.Translog;
 
@@ -33,25 +34,38 @@
  */
 public final class ResyncReplicationRequest extends ReplicatedWriteRequest<ResyncReplicationRequest> {
 
+    private long belowTermId;
+    private long trimmedAboveSeqNo;
     private Translog.Operation[] operations;
 
     ResyncReplicationRequest() {
         super();
     }
 
-    public ResyncReplicationRequest(final ShardId shardId, final Translog.Operation[] operations) {
+    public ResyncReplicationRequest(final ShardId shardId, final long belowTermId, final long trimmedAboveSeqNo,
+                                    final Translog.Operation[] operations) {
         super(shardId);
+        this.belowTermId = belowTermId;
+        this.trimmedAboveSeqNo = trimmedAboveSeqNo;
         this.operations = operations;
     }
 
+    public long getBelowTermId() {
+        return belowTermId;
+    }
+
+    public long getTrimmedAboveSeqNo() {
+        return trimmedAboveSeqNo;
+    }
+
     public Translog.Operation[] getOperations() {
         return operations;
     }
 
     @Override
     public void readFrom(final StreamInput in) throws IOException {
         assert Version.CURRENT.major <= 7;
-        if (in.getVersion().equals(Version.V_6_0_0)) {
+        if (in.getVersion().onOrBefore(Version.V_6_0_0)) {
             /*
              * Resync replication request serialization was broken in 6.0.0 due to the elements of the stream not being prefixed with a
              * byte indicating the type of the operation.
@@ -60,12 +74,22 @@ public void readFrom(final StreamInput in) throws IOException {
             throw new IllegalStateException("resync replication request serialization is broken in 6.0.0");
         }
         super.readFrom(in);
+        if (in.getVersion().onOrAfter(Version.V_6_4_0)) {
+            belowTermId = in.readVLong();
+            trimmedAboveSeqNo = in.readVLong();
+        } else {
+            trimmedAboveSeqNo = SequenceNumbers.UNASSIGNED_SEQ_NO;
+        }
         operations = in.readArray(Translog.Operation::readOperation, Translog.Operation[]::new);
     }
 
     @Override
     public void writeTo(final StreamOutput out) throws IOException {
         super.writeTo(out);
+        if (out.getVersion().onOrAfter(Version.V_6_4_0)) {
+            out.writeVLong(belowTermId);
+            out.writeVLong(trimmedAboveSeqNo);
+        }
         out.writeArray(Translog.Operation::writeOperation, operations);
     }
 
@@ -74,12 +98,14 @@ public boolean equals(final Object o) {
         if (this == o) return true;
         if (o == null || getClass() != o.getClass()) return false;
         final ResyncReplicationRequest that = (ResyncReplicationRequest) o;
-        return Arrays.equals(operations, that.operations);
+        return belowTermId == that.belowTermId
+            && trimmedAboveSeqNo == that.trimmedAboveSeqNo
+            && Arrays.equals(operations, that.operations);
     }
 
     @Override
     public int hashCode() {
-        return Arrays.hashCode(operations);
+        return Long.hashCode(belowTermId) + 31 * (Long.hashCode(trimmedAboveSeqNo) + 31 * Arrays.hashCode(operations));
     }
 
     @Override
@@ -88,6 +114,8 @@ public String toString() {
             "shardId=" + shardId +
             ", timeout=" + timeout +
             ", index='" + index + '\'' +
+            ", belowTermId=" + belowTermId +
+            ", trimmedAboveSeqNo=" + trimmedAboveSeqNo +
             ", ops=" + operations.length +
             "}";
     }

server/src/main/java/org/elasticsearch/action/resync/TransportResyncReplicationAction.java

@@ -135,6 +135,7 @@ public static Translog.Location performOnReplica(ResyncReplicationRequest reques
                 }
             }
         }
+        replica.trimTranslog(request.getBelowTermId(), request.getTrimmedAboveSeqNo());
         return location;
     }
 

server/src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -235,6 +235,13 @@ boolean isThrottled() {
      */
     public abstract boolean isThrottled();
 
+    /**
+     * Trims translog for terms below <code>belowTerm</code> and seq# above <code>aboveSeqNo</code>
+     */
+    public void trimTranslog(long belowTerm, long aboveSeqNo) throws IOException {
+        getTranslog().trim(belowTerm, aboveSeqNo);
+    }
+
     /** A Lock implementation that always allows the lock to be acquired */
     protected static final class NoOpLock implements Lock {
 

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -1197,6 +1197,10 @@ public void prepareForIndexRecovery() {
         assert currentEngineReference.get() == null;
     }
 
+    public void trimTranslog(long belowTerm, long aboveSeqNo) throws IOException {
+        getEngine().trimTranslog(belowTerm, aboveSeqNo);
+    }
+
     public Engine.Result applyTranslogOperation(Translog.Operation operation, Engine.Operation.Origin origin,
                                                 Consumer<Mapping> onMappingUpdate) throws IOException {
         final Engine.Result result;

server/src/main/java/org/elasticsearch/index/shard/PrimaryReplicaSyncer.java

@@ -35,6 +35,7 @@
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.index.seqno.SeqNoStats;
 import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.index.translog.Translog;
 import org.elasticsearch.tasks.Task;
@@ -83,6 +84,8 @@ public void resync(final IndexShard indexShard, final ActionListener<ResyncTask>
         ActionListener<ResyncTask> resyncListener = null;
         try {
             final long startingSeqNo = indexShard.getGlobalCheckpoint() + 1;
+            final SeqNoStats seqNoStats = indexShard.seqNoStats();
+            final long maxSeqNo = seqNoStats != null ? seqNoStats.getMaxSeqNo() : SequenceNumbers.UNASSIGNED_SEQ_NO;
             Translog.Snapshot snapshot = indexShard.newTranslogSnapshotFromMinSeqNo(startingSeqNo);
             resyncListener = new ActionListener<ResyncTask>() {
                 @Override
@@ -135,7 +138,7 @@ public synchronized Translog.Operation next() throws IOException {
                 }
             };
             resync(shardId, indexShard.routingEntry().allocationId().getId(), indexShard.getPrimaryTerm(), wrappedSnapshot,
-                startingSeqNo, resyncListener);
+                startingSeqNo, maxSeqNo, resyncListener);
         } catch (Exception e) {
             if (resyncListener != null) {
                 resyncListener.onFailure(e);
@@ -146,7 +149,7 @@ public synchronized Translog.Operation next() throws IOException {
     }
 
     private void resync(final ShardId shardId, final String primaryAllocationId, final long primaryTerm, final Translog.Snapshot snapshot,
-                        long startingSeqNo, ActionListener<ResyncTask> listener) {
+                        long startingSeqNo, long maxSeqNo, ActionListener<ResyncTask> listener) {
         ResyncRequest request = new ResyncRequest(shardId, primaryAllocationId);
         ResyncTask resyncTask = (ResyncTask) taskManager.register("transport", "resync", request); // it's not transport :-)
         ActionListener<Void> wrappedListener = new ActionListener<Void>() {
@@ -166,7 +169,7 @@ public void onFailure(Exception e) {
         };
         try {
             new SnapshotSender(logger, syncAction, resyncTask, shardId, primaryAllocationId, primaryTerm, snapshot, chunkSize.bytesAsInt(),
-                startingSeqNo, wrappedListener).run();
+                startingSeqNo, maxSeqNo, wrappedListener).run();
         } catch (Exception e) {
             wrappedListener.onFailure(e);
         }
@@ -186,14 +189,16 @@ static class SnapshotSender extends AbstractRunnable implements ActionListener<R
         private final ShardId shardId;
         private final Translog.Snapshot snapshot;
         private final long startingSeqNo;
+        private final long maxSeqNo;
         private final int chunkSizeInBytes;
         private final ActionListener<Void> listener;
+        private final AtomicInteger totalPackets = new AtomicInteger();
         private final AtomicInteger totalSentOps = new AtomicInteger();
         private final AtomicInteger totalSkippedOps = new AtomicInteger();
         private AtomicBoolean closed = new AtomicBoolean();
 
         SnapshotSender(Logger logger, SyncAction syncAction, ResyncTask task, ShardId shardId, String primaryAllocationId, long primaryTerm,
-                       Translog.Snapshot snapshot, int chunkSizeInBytes, long startingSeqNo, ActionListener<Void> listener) {
+                       Translog.Snapshot snapshot, int chunkSizeInBytes, long startingSeqNo, long maxSeqNo, ActionListener<Void> listener) {
             this.logger = logger;
             this.syncAction = syncAction;
             this.task = task;
@@ -203,6 +208,7 @@ static class SnapshotSender extends AbstractRunnable implements ActionListener<R
             this.snapshot = snapshot;
             this.chunkSizeInBytes = chunkSizeInBytes;
             this.startingSeqNo = startingSeqNo;
+            this.maxSeqNo = maxSeqNo;
             this.listener = listener;
             task.setTotalOperations(snapshot.totalOperations());
         }
@@ -248,11 +254,15 @@ protected void doRun() throws Exception {
                 }
             }
 
-            if (!operations.isEmpty()) {
+            final long trimmedAboveSeqNo = totalPackets.get() == 0 ? maxSeqNo : SequenceNumbers.UNASSIGNED_SEQ_NO;
+            // have to send sync request even in case of there are no operations to sync - have to sync trimmedAboveSeqNo at least
+            if (!operations.isEmpty() || totalPackets.get() == 0) {
                 task.setPhase("sending_ops");
-                ResyncReplicationRequest request = new ResyncReplicationRequest(shardId, operations.toArray(EMPTY_ARRAY));
+                ResyncReplicationRequest request =
+                    new ResyncReplicationRequest(shardId, primaryTerm, trimmedAboveSeqNo, operations.toArray(EMPTY_ARRAY));
                 logger.trace("{} sending batch of [{}][{}] (total sent: [{}], skipped: [{}])", shardId, operations.size(),
                     new ByteSizeValue(size), totalSentOps.get(), totalSkippedOps.get());
+                totalPackets.incrementAndGet();
                 syncAction.sync(request, task, primaryAllocationId, primaryTerm, this);
             } else if (closed.compareAndSet(false, true)) {
                 logger.trace("{} resync completed (total sent: [{}], skipped: [{}])", shardId, totalSentOps.get(), totalSkippedOps.get());

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

@@ -45,14 +45,29 @@ final class Checkpoint {
     final long maxSeqNo;
     final long globalCheckpoint;
     final long minTranslogGeneration;
+    final long trimmedAboveSeqNo;
 
     private static final int INITIAL_VERSION = 1; // start with 1, just to recognize there was some magic serialization logic before
-    private static final int CURRENT_VERSION = 2; // introduction of global checkpoints
+    private static final int VERSION_6_0_0 = 2; // introduction of global checkpoints
+    private static final int CURRENT_VERSION = 3; // introduction of global checkpoints
 
     private static final String CHECKPOINT_CODEC = "ckp";
 
-    // size of 6.0.0 checkpoint
+    // size of 6.4.0 checkpoint
+
     static final int FILE_SIZE = CodecUtil.headerLength(CHECKPOINT_CODEC)
+        + Integer.BYTES  // ops
+        + Long.BYTES // offset
+        + Long.BYTES // generation
+        + Long.BYTES // minimum sequence number, introduced in 6.0.0
+        + Long.BYTES // maximum sequence number, introduced in 6.0.0
+        + Long.BYTES // global checkpoint, introduced in 6.0.0
+        + Long.BYTES // minimum translog generation in the translog - introduced in 6.0.0
+        + Long.BYTES // maximum reachable (trimmed) sequence number, introduced in 6.4.0
+        + CodecUtil.footerLength();
+
+    // size of 6.0.0 checkpoint
+    static final int V2_FILE_SIZE = CodecUtil.headerLength(CHECKPOINT_CODEC)
         + Integer.BYTES  // ops
         + Long.BYTES // offset
         + Long.BYTES // generation
@@ -79,9 +94,12 @@ final class Checkpoint {
      * @param maxSeqNo         the current maximum sequence number of all operations in the translog
      * @param globalCheckpoint the last-known global checkpoint
      * @param minTranslogGeneration the minimum generation referenced by the translog at this moment.
+     * @param trimmedAboveSeqNo   the current maximum reachable (trimmed) sequence number of all operations in the translog
      */
-    Checkpoint(long offset, int numOps, long generation, long minSeqNo, long maxSeqNo, long globalCheckpoint, long minTranslogGeneration) {
+    Checkpoint(long offset, int numOps, long generation, long minSeqNo, long maxSeqNo, long globalCheckpoint,
+               long minTranslogGeneration, long trimmedAboveSeqNo) {
         assert minSeqNo <= maxSeqNo : "minSeqNo [" + minSeqNo + "] is higher than maxSeqNo [" + maxSeqNo + "]";
+        assert trimmedAboveSeqNo <= maxSeqNo : "trimmedAboveSeqNo [" + trimmedAboveSeqNo + "] is higher than maxSeqNo [" + maxSeqNo + "]";
         assert minTranslogGeneration <= generation :
             "minTranslogGen [" + minTranslogGeneration + "] is higher than generation [" + generation + "]";
         this.offset = offset;
@@ -91,6 +109,7 @@ final class Checkpoint {
         this.maxSeqNo = maxSeqNo;
         this.globalCheckpoint = globalCheckpoint;
         this.minTranslogGeneration = minTranslogGeneration;
+        this.trimmedAboveSeqNo = trimmedAboveSeqNo;
     }
 
     private void write(DataOutput out) throws IOException {
@@ -101,26 +120,49 @@ private void write(DataOutput out) throws IOException {
         out.writeLong(maxSeqNo);
         out.writeLong(globalCheckpoint);
         out.writeLong(minTranslogGeneration);
+        out.writeLong(trimmedAboveSeqNo);
     }
 
     static Checkpoint emptyTranslogCheckpoint(final long offset, final long generation, final long globalCheckpoint,
                                               long minTranslogGeneration) {
         final long minSeqNo = SequenceNumbers.NO_OPS_PERFORMED;
         final long maxSeqNo = SequenceNumbers.NO_OPS_PERFORMED;
-        return new Checkpoint(offset, 0, generation, minSeqNo, maxSeqNo, globalCheckpoint, minTranslogGeneration);
+        return new Checkpoint(offset, 0, generation, minSeqNo, maxSeqNo, globalCheckpoint, minTranslogGeneration, maxSeqNo);
+    }
+
+    static Checkpoint readCheckpointV6_4_0(final DataInput in) throws IOException {
+        final long offset = in.readLong();
+        final int numOps = in.readInt();
+        final long generation = in.readLong();
+        final long minSeqNo = in.readLong();
+        final long maxSeqNo = in.readLong();
+        final long globalCheckpoint = in.readLong();
+        final long minTranslogGeneration = in.readLong();
+        final long trimmedAboveSeqNo = in.readLong();
+        return new Checkpoint(offset, numOps, generation, minSeqNo, maxSeqNo, globalCheckpoint, minTranslogGeneration, trimmedAboveSeqNo);
     }
 
     static Checkpoint readCheckpointV6_0_0(final DataInput in) throws IOException {
-        return new Checkpoint(in.readLong(), in.readInt(), in.readLong(), in.readLong(), in.readLong(), in.readLong(), in.readLong());
+        final long offset = in.readLong();
+        final int numOps = in.readInt();
+        final long generation = in.readLong();
+        final long minSeqNo = in.readLong();
+        final long maxSeqNo = in.readLong();
+        final long globalCheckpoint = in.readLong();
+        final long minTranslogGeneration = in.readLong();
+        return new Checkpoint(offset, numOps, generation, minSeqNo, maxSeqNo, globalCheckpoint, minTranslogGeneration, maxSeqNo);
     }
 
     // reads a checksummed checkpoint introduced in ES 5.0.0
     static Checkpoint readCheckpointV5_0_0(final DataInput in) throws IOException {
+        final long offset = in.readLong();
+        final int numOps = in.readInt();
+        final long generation = in.readLong();
         final long minSeqNo = SequenceNumbers.NO_OPS_PERFORMED;
         final long maxSeqNo = SequenceNumbers.NO_OPS_PERFORMED;
         final long globalCheckpoint = SequenceNumbers.UNASSIGNED_SEQ_NO;
-        final long minTranslogGeneration = -1L;
-        return new Checkpoint(in.readLong(), in.readInt(), in.readLong(), minSeqNo, maxSeqNo, globalCheckpoint, minTranslogGeneration);
+        final long minTranslogGeneration = SequenceNumbers.UNASSIGNED_SEQ_NO;
+        return new Checkpoint(offset, numOps, generation, minSeqNo, maxSeqNo, globalCheckpoint, minTranslogGeneration, maxSeqNo);
     }
 
     @Override
@@ -133,6 +175,7 @@ public String toString() {
             ", maxSeqNo=" + maxSeqNo +
             ", globalCheckpoint=" + globalCheckpoint +
             ", minTranslogGeneration=" + minTranslogGeneration +
+            ", trimmedAboveSeqNo=" + trimmedAboveSeqNo +
             '}';
     }
 
@@ -145,10 +188,13 @@ public static Checkpoint read(Path path) throws IOException {
                 if (fileVersion == INITIAL_VERSION) {
                     assert indexInput.length() == V1_FILE_SIZE : indexInput.length();
                     return Checkpoint.readCheckpointV5_0_0(indexInput);
+                } else if (fileVersion == VERSION_6_0_0) {
+                    assert indexInput.length() == V2_FILE_SIZE : indexInput.length();
+                    return Checkpoint.readCheckpointV6_0_0(indexInput);
                 } else {
                     assert fileVersion == CURRENT_VERSION : fileVersion;
                     assert indexInput.length() == FILE_SIZE : indexInput.length();
-                    return Checkpoint.readCheckpointV6_0_0(indexInput);
+                    return Checkpoint.readCheckpointV6_4_0(indexInput);
                 }
             }
         }
@@ -164,7 +210,7 @@ public synchronized byte[] toByteArray() {
         };
         final String resourceDesc = "checkpoint(path=\"" + checkpointFile + "\", gen=" + checkpoint + ")";
         try (OutputStreamIndexOutput indexOutput =
-                 new OutputStreamIndexOutput(resourceDesc, checkpointFile.toString(), byteOutputStream, FILE_SIZE)) {
+                 new OutputStreamIndexOutput(resourceDesc, checkpointFile.toString(), byteOutputStream, V2_FILE_SIZE)) {
             CodecUtil.writeHeader(indexOutput, CHECKPOINT_CODEC, CURRENT_VERSION);
             checkpoint.write(indexOutput);
             CodecUtil.writeFooter(indexOutput);
@@ -196,7 +242,8 @@ public boolean equals(Object o) {
         if (generation != that.generation) return false;
         if (minSeqNo != that.minSeqNo) return false;
         if (maxSeqNo != that.maxSeqNo) return false;
-        return globalCheckpoint == that.globalCheckpoint;
+        if (globalCheckpoint != that.globalCheckpoint) return false;
+        return trimmedAboveSeqNo == that.trimmedAboveSeqNo;
     }
 
     @Override
@@ -207,6 +254,7 @@ public int hashCode() {
         result = 31 * result + Long.hashCode(minSeqNo);
         result = 31 * result + Long.hashCode(maxSeqNo);
         result = 31 * result + Long.hashCode(globalCheckpoint);
+        result = 31 * result + Long.hashCode(trimmedAboveSeqNo);
         return result;
     }