improvements to iterating over commit data (and better safety guarantees)

Author: Ali Beyad

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

@@ -74,6 +74,7 @@
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -282,7 +283,7 @@ private Translog openTranslog(EngineConfig engineConfig, IndexWriter writer) thr
             boolean success = false;
             try {
                 commitIndexWriter(writer, translog, openMode == EngineConfig.OpenMode.OPEN_INDEX_CREATE_TRANSLOG
-                    ? writer.getCommitData().get(SYNC_COMMIT_ID) : null);
+                    ? commitDataAsMap(writer).get(SYNC_COMMIT_ID) : null);
                 success = true;
             } finally {
                 if (success == false) {
@@ -307,7 +308,7 @@ public Translog getTranslog() {
     private Translog.TranslogGeneration loadTranslogIdFromCommit(IndexWriter writer) throws IOException {
         // commit on a just opened writer will commit even if there are no changes done to it
         // we rely on that for the commit data translog id key
-        final Map<String, String> commitUserData = writer.getCommitData();
+        final Map<String, String> commitUserData = commitDataAsMap(writer);
         if (commitUserData.containsKey("translog_id")) {
             assert commitUserData.containsKey(Translog.TRANSLOG_UUID_KEY) == false : "legacy commit contains translog UUID";
             return new Translog.TranslogGeneration(null, Long.parseLong(commitUserData.get("translog_id")));
@@ -322,6 +323,11 @@ private Translog.TranslogGeneration loadTranslogIdFromCommit(IndexWriter writer)
         return null;
     }
 
+    // package private for testing
+    SeqNoStats loadSeqNoStatsFromCommit() throws IOException {
+        return loadSeqNoStatsFromCommit(indexWriter);
+    }
+
     private SeqNoStats loadSeqNoStatsFromCommit(IndexWriter writer) throws IOException {
         long maxSeqNo = SequenceNumbersService.NO_OPS_PERFORMED;
         long localCheckpoint = SequenceNumbersService.NO_OPS_PERFORMED;
@@ -1308,40 +1314,52 @@ private void commitIndexWriter(IndexWriter writer, Translog translog, String syn
         ensureCanFlush();
         try {
             Translog.TranslogGeneration translogGeneration = translog.getGeneration();
-            final Map<String, String> commitData = new HashMap<>(5);
 
-            commitData.put(Translog.TRANSLOG_GENERATION_KEY, Long.toString(translogGeneration.translogFileGeneration));
-            commitData.put(Translog.TRANSLOG_UUID_KEY, translogGeneration.translogUUID);
+            final String translogFileGen = Long.toString(translogGeneration.translogFileGeneration);
+            final String translogUUID = translogGeneration.translogUUID;
+            final String localCheckpoint = Long.toString(seqNoService().getLocalCheckpoint());
+            final String globalCheckpoint = Long.toString(seqNoService().getGlobalCheckpoint());
 
-            commitData.put(LOCAL_CHECKPOINT_KEY, Long.toString(seqNoService().getLocalCheckpoint()));
-            commitData.put(GLOBAL_CHECKPOINT_KEY, Long.toString(seqNoService().getGlobalCheckpoint()));
+            writer.setLiveCommitData(new Iterable<Map.Entry<String, String>>() {
+                // save the max seq no the first time its computed, so subsequent iterations don't recompute,
+                // potentially getting a different value
+                private String computedMaxSeqNoEntry = null;
 
-            if (syncId != null) {
-                commitData.put(Engine.SYNC_COMMIT_ID, syncId);
-            }
+                @Override
+                public Iterator<Map.Entry<String, String>> iterator() {
+                    /**
+                     * The user data captured above (e.g. local/global checkpoints) contains data that must be evaluated
+                     * *before* Lucene flushes segments, including the local and global checkpoints amongst other values.
+                     * The maximum sequence number is different - we never want the maximum sequence number to be
+                     * less than the last sequence number to go into a Lucene commit, otherwise we run the risk
+                     * of re-using a sequence number for two different documents when restoring from this commit
+                     * point and subsequently writing new documents to the index.  Since we only know which Lucene
+                     * documents made it into the final commit after the {@link IndexWriter#commit()} call flushes
+                     * all documents, we defer computation of the max_seq_no to the time of invocation of the commit
+                     * data iterator (which occurs after all documents have been flushed to Lucene).
+                     */
+                    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(GLOBAL_CHECKPOINT_KEY, globalCheckpoint);
+                    if (syncId != null) {
+                        commitData.put(Engine.SYNC_COMMIT_ID, syncId);
+                    }
+                    if (computedMaxSeqNoEntry == null) {
+                        // evaluated once at the time of the first invocation of this method
+                        computedMaxSeqNoEntry = Long.toString(seqNoService().getMaxSeqNo());
+                    }
+                    commitData.put(MAX_SEQ_NO, computedMaxSeqNoEntry);
+                    return commitData.entrySet().iterator();
+                }
+            });
+
+            writer.commit();
 
             if (logger.isTraceEnabled()) {
-                logger.trace("committing writer with commit data (max_seq_no excluded) [{}]", commitData);
+                logger.trace("committed writer with commit data [{}]", commitDataAsMap(writer));
             }
-
-            indexWriter.setLiveCommitData(() -> {
-                /**
-                 * The user data in the commitData map contains data that must be evaluated *before*
-                 * Lucene flushes segments, including the local and global checkpoints amongst other values.
-                 * The maximum sequence number is different - we never want the maximum sequence number to be
-                 * less than the last sequence number to go into a Lucene commit, otherwise we run the risk
-                 * of re-using a sequence number for two different documents when restoring from this commit
-                 * point and subsequently writing new documents to the index.  Since we only know which Lucene
-                 * documents made it into the final commit after the {@link IndexWriter#commit()} call flushes
-                 * all documents, we defer computation of the max_seq_no to the time of invocation of the commit
-                 * data iterator (which occurs after all documents have been flushed to Lucene).
-                 */
-                final Map<String, String> deferredCommitData = new HashMap<>(commitData.size() + 1);
-                deferredCommitData.putAll(commitData);
-                deferredCommitData.put(MAX_SEQ_NO, Long.toString(seqNoService().getMaxSeqNo()));
-                return deferredCommitData.entrySet().iterator();
-            });
-            writer.commit();
         } catch (Exception ex) {
             try {
                 failEngine("lucene commit failed", ex);
@@ -1443,4 +1461,15 @@ boolean indexWriterHasDeletions() {
     public boolean isRecovering() {
         return pendingTranslogRecovery.get();
     }
+
+    /**
+     * Gets the commit data from {@link IndexWriter} as a map.
+     */
+    private static Map<String, String> commitDataAsMap(final IndexWriter indexWriter) {
+        Map<String, String> commitData = new HashMap<>(6);
+        for (Map.Entry<String, String> entry : indexWriter.getLiveCommitData()) {
+            commitData.put(entry.getKey(), entry.getValue());
+        }
+        return commitData;
+    }
 }

core/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java

@@ -138,6 +138,7 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.everyItem;
 import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasKey;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.notNullValue;
@@ -1646,18 +1647,33 @@ public void testSeqNoAndCheckpoints() throws IOException {
                     SequenceNumbersService.UNASSIGNED_SEQ_NO,
                     rarely() ? 100 : Versions.MATCH_ANY, VersionType.INTERNAL,
                     PRIMARY, 0, -1, false);
+                boolean versionConflict = false;
                 try {
                     initialEngine.index(index);
                     primarySeqNo++;
                 } catch (VersionConflictEngineException e) {
-
+                    versionConflict = true;
                 }
 
                 replicaLocalCheckpoint =
                     rarely() ? replicaLocalCheckpoint : randomIntBetween(Math.toIntExact(replicaLocalCheckpoint), Math.toIntExact(primarySeqNo));
                 initialEngine.seqNoService().updateLocalCheckpointForShard("primary", initialEngine.seqNoService().getLocalCheckpoint());
                 initialEngine.seqNoService().updateLocalCheckpointForShard("replica", replicaLocalCheckpoint);
 
+                // make sure the max seq no in the latest commit hasn't advanced due to more documents having been added;
+                // the first time the commit data iterable gets an iterator, the max seq no from that point in time should
+                // remain from any subsequent call to IndexWriter#getLiveCommitData unless the commit data is overridden by a
+                // subsequent call to IndexWriter#setLiveCommitData.
+                if (initialEngine.seqNoService().getMaxSeqNo() != SequenceNumbersService.NO_OPS_PERFORMED) {
+                    assertThat(
+                        initialEngine.seqNoService().getMaxSeqNo(),
+                        // its possible that right after a commit, a version conflict exception happened so the max seq no was not updated,
+                        // so here we check greater than or equal to
+                        versionConflict ? greaterThanOrEqualTo(initialEngine.loadSeqNoStatsFromCommit().getMaxSeqNo()) :
+                                          greaterThan(initialEngine.loadSeqNoStatsFromCommit().getMaxSeqNo())
+                    );
+                }
+
                 if (rarely()) {
                     localCheckpoint = primarySeqNo;
                     maxSeqNo = primarySeqNo;