Introduce sequence number-based recovery

This commit introduces sequence-number-based recovery. When a replica
has fallen out of sync, rather than performing a file-based recovery we
first attempt to replay operations since the last local checkpoint on
the replica. To do this, at the start of recovery the replica tells the
primary what its local checkpoint is. The primary will then wait for all
operations between that local checkpoint and the current maximum
sequence number to complete; this is to ensure that there are no gaps in
the operations that will be replayed from the primary to the
replica. This is a best-effort attempt as we currently have no
guarantees on the primary that these operations will be available; if we
are not able to replay all operations in the desired range, we just
fallback to file-based recovery. Later work will strengthen the
guarantees.

Author: jasontedor

buildSrc/src/main/resources/checkstyle_suppressions.xml

@@ -424,7 +424,6 @@
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]indices[/\\]recovery[/\\]RecoverySettings.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]indices[/\\]recovery[/\\]PeerRecoverySourceService.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]indices[/\\]recovery[/\\]RecoveryState.java" checks="LineLength" />
-  <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]indices[/\\]recovery[/\\]StartRecoveryRequest.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]indices[/\\]store[/\\]IndicesStore.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]indices[/\\]store[/\\]TransportNodesListShardStoreMetaData.java" checks="LineLength" />
   <suppress files="core[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]indices[/\\]ttl[/\\]IndicesTTLService.java" checks="LineLength" />

core/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java

@@ -36,7 +36,7 @@
 import org.elasticsearch.index.fielddata.IndexFieldDataService;
 import org.elasticsearch.index.mapper.FieldMapper;
 import org.elasticsearch.index.mapper.MapperService;
-import org.elasticsearch.index.seqno.LocalCheckpointService;
+import org.elasticsearch.index.seqno.LocalCheckpointTracker;
 import org.elasticsearch.index.similarity.SimilarityService;
 import org.elasticsearch.index.store.FsDirectoryService;
 import org.elasticsearch.index.store.Store;
@@ -113,7 +113,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
         IndexSettings.ALLOW_UNMAPPED,
         IndexSettings.INDEX_CHECK_ON_STARTUP,
         IndexSettings.INDEX_SEQ_NO_CHECKPOINT_SYNC_INTERVAL,
-        LocalCheckpointService.SETTINGS_BIT_ARRAYS_SIZE,
+        LocalCheckpointTracker.SETTINGS_BIT_ARRAYS_SIZE,
         IndexSettings.MAX_REFRESH_LISTENERS_PER_SHARD,
         IndexSettings.MAX_SLICES_PER_SCROLL,
         ShardsLimitAllocationDecider.INDEX_TOTAL_SHARDS_PER_NODE_SETTING,

core/src/main/java/org/elasticsearch/common/util/concurrent/AbstractRefCounted.java

@@ -16,6 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+
 package org.elasticsearch.common.util.concurrent;
 
 import org.apache.lucene.store.AlreadyClosedException;

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

@@ -361,7 +361,7 @@ public Operation.TYPE getOperationType() {
 
         void setTranslogLocation(Translog.Location translogLocation) {
             if (freeze.get() == null) {
-                assert failure == null : "failure has to be null to set translog location";
+                assert failure == null || translogLocation == null: "failure has to be null to set translog location";
                 this.translogLocation = translogLocation;
             } else {
                 throw new IllegalStateException("result is already frozen");
@@ -379,6 +379,7 @@ void setTook(long took) {
         void freeze() {
             freeze.set(true);
         }
+
     }
 
     public static class IndexResult extends Result {

core/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -63,6 +63,7 @@
 import org.elasticsearch.index.merge.MergeStats;
 import org.elasticsearch.index.merge.OnGoingMerge;
 import org.elasticsearch.index.seqno.SeqNoStats;
+import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.index.seqno.SequenceNumbersService;
 import org.elasticsearch.index.shard.ElasticsearchMergePolicy;
 import org.elasticsearch.index.shard.ShardId;
@@ -119,8 +120,6 @@ public class InternalEngine extends Engine {
     private final IndexThrottle throttle;
 
     private final SequenceNumbersService seqNoService;
-    static final String LOCAL_CHECKPOINT_KEY = "local_checkpoint";
-    static final String MAX_SEQ_NO = "max_seq_no";
 
     // How many callers are currently requesting index throttling.  Currently there are only two situations where we do this: when merges
     // are falling behind and when writing indexing buffer to disk is too slow.  When this is 0, there is no throttling, else we throttling
@@ -365,7 +364,7 @@ private Translog.TranslogGeneration loadTranslogIdFromCommit(IndexWriter writer)
     private static SeqNoStats loadSeqNoStatsFromLuceneAndTranslog(
         final TranslogConfig translogConfig,
         final IndexWriter indexWriter) throws IOException {
-        long globalCheckpoint = Translog.readGlobalCheckpoint(translogConfig.getTranslogPath());
+        final long globalCheckpoint = Translog.readGlobalCheckpoint(translogConfig.getTranslogPath());
         return loadSeqNoStatsFromLucene(globalCheckpoint, indexWriter);
     }
 
@@ -378,20 +377,7 @@ private static SeqNoStats loadSeqNoStatsFromLuceneAndTranslog(
      * @return the sequence number stats
      */
     private static SeqNoStats loadSeqNoStatsFromLucene(final long globalCheckpoint, final IndexWriter indexWriter) {
-        long maxSeqNo = SequenceNumbersService.NO_OPS_PERFORMED;
-        long localCheckpoint = SequenceNumbersService.NO_OPS_PERFORMED;
-        for (Map.Entry<String, String> entry : indexWriter.getLiveCommitData()) {
-            final String key = entry.getKey();
-            if (key.equals(LOCAL_CHECKPOINT_KEY)) {
-                assert localCheckpoint == SequenceNumbersService.NO_OPS_PERFORMED;
-                localCheckpoint = Long.parseLong(entry.getValue());
-            } else if (key.equals(MAX_SEQ_NO)) {
-                assert maxSeqNo == SequenceNumbersService.NO_OPS_PERFORMED : localCheckpoint;
-                maxSeqNo = Long.parseLong(entry.getValue());
-            }
-        }
-
-        return new SeqNoStats(maxSeqNo, localCheckpoint, globalCheckpoint);
+        return SequenceNumbers.loadSeqNoStatsFromLuceneCommit(globalCheckpoint, indexWriter.getLiveCommitData());
     }
 
     private SearcherManager createSearcherManager() throws EngineException {
@@ -684,13 +670,20 @@ private IndexResult innerIndex(Index index) throws IOException {
             final IndexResult indexResult;
             if (checkVersionConflictResult.isPresent()) {
                 indexResult = checkVersionConflictResult.get();
+                // norelease: this is not correct as this does not force an fsync, and we need to handle failures including replication
+                if (indexResult.hasFailure()) {
+                    location = null;
+                } else {
+                    final Translog.NoOp operation = new Translog.NoOp(seqNo, index.primaryTerm(), "version conflict during recovery");
+                    location = index.origin() != Operation.Origin.LOCAL_TRANSLOG_RECOVERY ? translog.add(operation) : null;
+                }
             } else {
                 // no version conflict
                 if (index.origin() == Operation.Origin.PRIMARY) {
                     seqNo = seqNoService().generateSeqNo();
                 }
 
-                /**
+                /*
                  * Update the document's sequence number and primary term; the sequence number here is derived here from either the sequence
                  * number service if this is on the primary, or the existing document's sequence number if this is on the replica. The
                  * primary term here has already been set, see IndexShard#prepareIndex where the Engine$Index operation is created.
@@ -707,12 +700,11 @@ private IndexResult innerIndex(Index index) throws IOException {
                     update(index.uid(), index.docs(), indexWriter);
                 }
                 indexResult = new IndexResult(updatedVersion, seqNo, deleted);
-                location = index.origin() != Operation.Origin.LOCAL_TRANSLOG_RECOVERY
-                    ? translog.add(new Translog.Index(index, indexResult))
-                    : null;
                 versionMap.putUnderLock(index.uid().bytes(), new VersionValue(updatedVersion));
-                indexResult.setTranslogLocation(location);
+                final Translog.Index operation = new Translog.Index(index, indexResult);
+                location = index.origin() != Operation.Origin.LOCAL_TRANSLOG_RECOVERY ? translog.add(operation) : null;
             }
+            indexResult.setTranslogLocation(location);
             indexResult.setTook(System.nanoTime() - index.startTime());
             indexResult.freeze();
             return indexResult;
@@ -816,21 +808,26 @@ private DeleteResult innerDelete(Delete delete) throws IOException {
             final DeleteResult deleteResult;
             if (result.isPresent()) {
                 deleteResult = result.get();
+                // norelease: this is not correct as this does not force an fsync, and we need to handle failures including replication
+                if (deleteResult.hasFailure()) {
+                    location = null;
+                } else {
+                    final Translog.NoOp operation = new Translog.NoOp(seqNo, delete.primaryTerm(), "version conflict during recovery");
+                    location = delete.origin() != Operation.Origin.LOCAL_TRANSLOG_RECOVERY ? translog.add(operation) : null;
+                }
             } else {
                 if (delete.origin() == Operation.Origin.PRIMARY) {
                     seqNo = seqNoService().generateSeqNo();
                 }
-
                 updatedVersion = delete.versionType().updateVersion(currentVersion, expectedVersion);
                 found = deleteIfFound(delete.uid(), currentVersion, deleted, versionValue);
                 deleteResult = new DeleteResult(updatedVersion, seqNo, found);
-                location = delete.origin() != Operation.Origin.LOCAL_TRANSLOG_RECOVERY
-                    ? translog.add(new Translog.Delete(delete, deleteResult))
-                    : null;
                 versionMap.putUnderLock(delete.uid().bytes(),
                     new DeleteVersionValue(updatedVersion, engineConfig.getThreadPool().estimatedTimeInMillis()));
-                deleteResult.setTranslogLocation(location);
+                final Translog.Delete operation = new Translog.Delete(delete, deleteResult);
+                location = delete.origin() != Operation.Origin.LOCAL_TRANSLOG_RECOVERY ? translog.add(operation) : null;
             }
+            deleteResult.setTranslogLocation(location);
             deleteResult.setTook(System.nanoTime() - delete.startTime());
             deleteResult.freeze();
             return deleteResult;
@@ -1552,11 +1549,11 @@ private void commitIndexWriter(IndexWriter writer, Translog translog, String syn
                 final Map<String, String> commitData = new HashMap<>(6);
                 commitData.put(Translog.TRANSLOG_GENERATION_KEY, translogFileGen);
                 commitData.put(Translog.TRANSLOG_UUID_KEY, translogUUID);
-                commitData.put(LOCAL_CHECKPOINT_KEY, localCheckpoint);
+                commitData.put(SequenceNumbers.LOCAL_CHECKPOINT_KEY, localCheckpoint);
                 if (syncId != null) {
                     commitData.put(Engine.SYNC_COMMIT_ID, syncId);
                 }
-                commitData.put(MAX_SEQ_NO, Long.toString(seqNoService().getMaxSeqNo()));
+                commitData.put(SequenceNumbers.MAX_SEQ_NO, Long.toString(seqNoService().getMaxSeqNo()));
                 if (logger.isTraceEnabled()) {
                     logger.trace("committing writer with commit data [{}]", commitData);
                 }

core/src/main/java/org/elasticsearch/index/seqno/GlobalCheckpointService.java renamed to core/src/main/java/org/elasticsearch/index/seqno/GlobalCheckpointTracker.java

@@ -22,25 +22,25 @@
 import com.carrotsearch.hppc.ObjectLongHashMap;
 import com.carrotsearch.hppc.ObjectLongMap;
 import com.carrotsearch.hppc.cursors.ObjectLongCursor;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.index.IndexSettings;
-import org.elasticsearch.index.shard.AbstractIndexShardComponent;
-import org.elasticsearch.index.shard.ShardId;
 
 import java.util.HashSet;
+import java.util.Locale;
 import java.util.Set;
 
 import static org.elasticsearch.index.seqno.SequenceNumbersService.UNASSIGNED_SEQ_NO;
 
 /**
- * A shard component that is responsible of tracking the global checkpoint. The global checkpoint is the highest sequence number for which
- * all lower (or equal) sequence number have been processed on all shards that are currently active. Since shards count as "active" when the
- * master starts them, and before this primary shard has been notified of this fact, we also include shards that have completed recovery.
- * These shards have received all old operations via the recovery mechanism and are kept up to date by the various replications actions.
- * The set of shards that are taken into account for the global checkpoint calculation are called the "in-sync shards".
+ * This class is responsible of tracking the global checkpoint. The global checkpoint is the highest sequence number for which all lower (or
+ * equal) sequence number have been processed on all shards that are currently active. Since shards count as "active" when the master starts
+ * them, and before this primary shard has been notified of this fact, we also include shards that have completed recovery. These shards
+ * have received all old operations via the recovery mechanism and are kept up to date by the various replications actions. The set of
+ * shards that are taken into account for the global checkpoint calculation are called the "in-sync shards".
  * <p>
  * The global checkpoint is maintained by the primary shard and is replicated to all the replicas (via {@link GlobalCheckpointSyncAction}).
  */
-public class GlobalCheckpointService extends AbstractIndexShardComponent {
+public class GlobalCheckpointTracker {
 
     /*
      * This map holds the last known local checkpoint for every active shard and initializing shard copies that has been brought up to speed
@@ -62,20 +62,22 @@ public class GlobalCheckpointService extends AbstractIndexShardComponent {
      */
     private long globalCheckpoint;
 
+    private final Logger logger;
+
     /**
-     * Initialize the global checkpoint service. The specified global checkpoint should be set to the last known global checkpoint for this
-     * shard, or {@link SequenceNumbersService#UNASSIGNED_SEQ_NO}.
+     * Initialize the global checkpoint service. The specified global checkpoint should be set to the last known global checkpoint, or
+     * {@link SequenceNumbersService#UNASSIGNED_SEQ_NO}.
      *
-     * @param shardId          the shard this service is tracking local checkpoints for
      * @param indexSettings    the index settings
      * @param globalCheckpoint the last known global checkpoint for this shard, or {@link SequenceNumbersService#UNASSIGNED_SEQ_NO}
+     * @param logger           a component logger
      */
-    GlobalCheckpointService(final ShardId shardId, final IndexSettings indexSettings, final long globalCheckpoint) {
-        super(shardId, indexSettings);
+    GlobalCheckpointTracker(final IndexSettings indexSettings, final long globalCheckpoint, final Logger logger) {
         assert globalCheckpoint >= UNASSIGNED_SEQ_NO : "illegal initial global checkpoint: " + globalCheckpoint;
         inSyncLocalCheckpoints = new ObjectLongHashMap<>(1 + indexSettings.getNumberOfReplicas());
         assignedAllocationIds = new HashSet<>(1 + indexSettings.getNumberOfReplicas());
         this.globalCheckpoint = globalCheckpoint;
+        this.logger = logger;
     }
 
     /**
@@ -127,8 +129,9 @@ synchronized boolean updateCheckpointOnPrimary() {
             minCheckpoint = Math.min(cp.value, minCheckpoint);
         }
         if (minCheckpoint < globalCheckpoint) {
-            throw new IllegalStateException(shardId + " new global checkpoint [" + minCheckpoint
-                + "] is lower than previous one [" + globalCheckpoint + "]");
+            final String message =
+                String.format(Locale.ROOT, "new global checkpoint [%d] is lower than previous one [%d]", minCheckpoint, globalCheckpoint);
+            throw new IllegalStateException(message);
         }
         if (globalCheckpoint != minCheckpoint) {
             logger.trace("global checkpoint updated to [{}]", minCheckpoint);