Add global checkpoint to translog checkpoints

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

@@ -167,9 +167,7 @@ public InternalEngine(EngineConfig engineConfig) throws EngineException {
                                 engineConfig.getIndexSettings(),
                                 seqNoStats.getMaxSeqNo(),
                                 seqNoStats.getLocalCheckpoint(),
-                                seqNoStats.getGlobalCheckpoint(),
-                                this::onGlobalCheckpointUpdate
-                            );
+                                seqNoStats.getGlobalCheckpoint());
                 indexWriter = writer;
                 translog = openTranslog(engineConfig, writer);
                 assert translog.getGeneration() != null;
@@ -360,19 +358,6 @@ private static SeqNoStats loadSeqNoStats(final EngineConfig engineConfig, final
         return new SeqNoStats(maxSeqNo, localCheckpoint, globalCheckpoint);
     }
 
-    /**
-     * Sync the translog after the global checkpoint is updated.
-     */
-    void onGlobalCheckpointUpdate() {
-        try (ReleasableLock ignored = readLock.acquire()) {
-            ensureOpen();
-            translog.sync();
-        } catch (final IOException e) {
-            maybeFailEngine("on global checkpoint update", e);
-            throw new EngineException(shardId, "failed on global checkpoint update", e);
-        }
-    }
-
     private SearcherManager createSearcherManager() throws EngineException {
         boolean success = false;
         SearcherManager searcherManager = null;

core/src/main/java/org/elasticsearch/index/seqno/GlobalCheckpointSyncAction.java

@@ -41,6 +41,8 @@
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.io.UnsupportedEncodingException;
 
 public class GlobalCheckpointSyncAction extends TransportReplicationAction<GlobalCheckpointSyncAction.PrimaryRequest,
     GlobalCheckpointSyncAction.ReplicaRequest, ReplicationResponse> {
@@ -68,6 +70,7 @@ protected PrimaryResult shardOperationOnPrimary(PrimaryRequest request) {
         IndexService indexService = indicesService.indexServiceSafe(request.shardId().getIndex());
         IndexShard indexShard = indexService.getShard(request.shardId().id());
         long checkpoint = indexShard.getGlobalCheckpoint();
+        syncTranslog(indexShard);
         return new PrimaryResult(new ReplicaRequest(request, checkpoint), new ReplicationResponse());
     }
 
@@ -76,9 +79,18 @@ protected ReplicaResult shardOperationOnReplica(ReplicaRequest request) {
         IndexService indexService = indicesService.indexServiceSafe(request.shardId().getIndex());
         IndexShard indexShard = indexService.getShard(request.shardId().id());
         indexShard.updateGlobalCheckpointOnReplica(request.checkpoint);
+        syncTranslog(indexShard);
         return new ReplicaResult();
     }
 
+    private void syncTranslog(final IndexShard indexShard) {
+        try {
+            indexShard.getTranslog().sync();
+        } catch (final IOException e) {
+            throw new UncheckedIOException("failed to sync translog after updating global checkpoint for shard " + indexShard.shardId(), e);
+        }
+    }
+
     public void updateCheckpointForShard(ShardId shardId) {
         execute(new PrimaryRequest(shardId), new ActionListener<ReplicationResponse>() {
             @Override
@@ -135,4 +147,5 @@ public long getCheckpoint() {
             return checkpoint;
         }
     }
+
 }

core/src/main/java/org/elasticsearch/index/seqno/SequenceNumbersService.java

@@ -38,7 +38,6 @@ public class SequenceNumbersService extends AbstractIndexShardComponent {
 
     final LocalCheckpointService localCheckpointService;
     final GlobalCheckpointService globalCheckpointService;
-    private final Runnable onGlobalCheckpointUpdate;
 
     /**
      * Initialize the sequence number service. The {@code maxSeqNo} should be set to the last sequence number assigned by this shard, or
@@ -51,19 +50,16 @@ public class SequenceNumbersService extends AbstractIndexShardComponent {
      * @param maxSeqNo                 the last sequence number assigned by this shard, or {@link SequenceNumbersService#NO_OPS_PERFORMED}
      * @param localCheckpoint          the last known local checkpoint for this shard, or {@link SequenceNumbersService#NO_OPS_PERFORMED}
      * @param globalCheckpoint         the last known global checkpoint for this shard, or {@link SequenceNumbersService#UNASSIGNED_SEQ_NO}
-     * @param onGlobalCheckpointUpdate invoked when the global checkpoint is updated
      */
     public SequenceNumbersService(
         final ShardId shardId,
         final IndexSettings indexSettings,
         final long maxSeqNo,
         final long localCheckpoint,
-        final long globalCheckpoint,
-        final Runnable onGlobalCheckpointUpdate) {
+        final long globalCheckpoint) {
         super(shardId, indexSettings);
         localCheckpointService = new LocalCheckpointService(shardId, indexSettings, maxSeqNo, localCheckpoint);
         globalCheckpointService = new GlobalCheckpointService(shardId, indexSettings, globalCheckpoint);
-        this.onGlobalCheckpointUpdate = onGlobalCheckpointUpdate;
     }
 
     /**
@@ -123,12 +119,21 @@ public long getGlobalCheckpoint() {
         return globalCheckpointService.getCheckpoint();
     }
 
+    /**
+     * Scans through the currently known local checkpoint and updates the global checkpoint accordingly.
+     *
+     * @return true if the checkpoint has been updated or if it can not be updated since one of the local checkpoints
+     * of one of the active allocations is not known.
+     */
+    public boolean updateGlobalCheckpointOnPrimary() {
+        return globalCheckpointService.updateCheckpointOnPrimary();
+    }
+
     /**
      * updates the global checkpoint on a replica shard (after it has been updated by the primary).
      */
     public void updateGlobalCheckpointOnReplica(long checkpoint) {
         globalCheckpointService.updateCheckpointOnReplica(checkpoint);
-        onGlobalCheckpointUpdate.run();
     }
 
     /**
@@ -142,17 +147,4 @@ public void updateAllocationIdsFromMaster(Set<String> activeAllocationIds, Set<S
         globalCheckpointService.updateAllocationIdsFromMaster(activeAllocationIds, initializingAllocationIds);
     }
 
-    /**
-     * Scans through the currently known local checkpoint and updates the global checkpoint accordingly.
-     *
-     * @return true if the checkpoint has been updated or if it can not be updated since one of the local checkpoints
-     * of one of the active allocations is not known.
-     */
-    public boolean updateGlobalCheckpointOnPrimary() {
-        final boolean maybeUpdateGlobalCheckpoint = globalCheckpointService.updateCheckpointOnPrimary();
-        if (maybeUpdateGlobalCheckpoint) {
-            onGlobalCheckpointUpdate.run();
-        }
-        return maybeUpdateGlobalCheckpoint;
-    }
 }

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

@@ -558,7 +558,7 @@ public boolean ensureSynced(Location location) throws IOException {
 
     /**
      * Ensures that all locations in the given stream have been synced / written to the underlying storage.
-     * This method allows for internal optimization to minimize the amout of fsync operations if multiple
+     * This method allows for internal optimization to minimize the amount of fsync operations if multiple
      * locations must be synced.
      *
      * @return Returns <code>true</code> iff this call caused an actual sync operation otherwise <code>false</code>

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

@@ -588,8 +588,7 @@ public SequenceNumbersService seqNoService() {
                         this.config().getIndexSettings(),
                         maxSeqNo.get(),
                         localCheckpoint.get(),
-                        globalCheckpoint.get(),
-                        () -> {});
+                        globalCheckpoint.get());
                 }
             };
             CommitStats stats1 = engine.commitStats();
@@ -1712,6 +1711,7 @@ public void testSeqNoAndCheckpoints() throws IOException {
             assertThat(
                 Long.parseLong(initialEngine.commitStats().getUserData().get(InternalEngine.LOCAL_CHECKPOINT_KEY)),
                 equalTo(localCheckpoint));
+            initialEngine.getTranslog().sync(); // to guarantee the global checkpoint is written to the translog checkpoint
             assertThat(
                 initialEngine.getTranslog().getLastSyncedGlobalCheckpoint(),
                 equalTo(globalCheckpoint));

core/src/test/java/org/elasticsearch/index/seqno/GlobalCheckpointSyncActionTests.java

@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache license, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the license for the specific language governing permissions and
+ * limitations under the license.
+ */
+
+package org.elasticsearch.index.seqno;
+
+import org.apache.lucene.util.IOUtils;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.cluster.action.shard.ShardStateAction;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.index.IndexService;
+import org.elasticsearch.index.shard.IndexShard;
+import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.index.translog.Translog;
+import org.elasticsearch.indices.IndicesService;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.transport.CapturingTransport;
+import org.elasticsearch.threadpool.TestThreadPool;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.Transport;
+import org.elasticsearch.transport.TransportService;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Collections;
+import java.util.HashSet;
+
+import static org.elasticsearch.mock.orig.Mockito.when;
+import static org.elasticsearch.test.ClusterServiceUtils.createClusterService;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+public class GlobalCheckpointSyncActionTests extends ESTestCase {
+
+    private ThreadPool threadPool;
+    private Transport transport;
+    private ClusterService clusterService;
+    private TransportService transportService;
+    private ShardStateAction shardStateAction;
+
+    public void setUp() throws Exception {
+        super.setUp();
+        threadPool = new TestThreadPool(getClass().getName());
+        transport = new CapturingTransport();
+        clusterService = createClusterService(threadPool);
+        transportService = new TransportService(clusterService.getSettings(), transport, threadPool,
+            TransportService.NOOP_TRANSPORT_INTERCEPTOR);
+        transportService.start();
+        transportService.acceptIncomingRequests();
+        shardStateAction = new ShardStateAction(Settings.EMPTY, clusterService, transportService, null, null, threadPool);
+    }
+
+    public void tearDown() throws Exception {
+        try {
+            IOUtils.close(transportService, clusterService, transport);
+        } finally {
+            terminate(threadPool);
+        }
+        super.tearDown();
+    }
+
+    public void testTranslogSyncAfterGlobalCheckpointSync() throws IOException {
+        final IndicesService indicesService = mock(IndicesService.class);
+
+        final Index index = new Index("index", "uuid");
+        final IndexService indexService = mock(IndexService.class);
+        when(indicesService.indexServiceSafe(index)).thenReturn(indexService);
+
+        final int id = randomIntBetween(0, 4);
+        final IndexShard indexShard = mock(IndexShard.class);
+        when(indexService.getShard(id)).thenReturn(indexShard);
+
+        final Translog translog = mock(Translog.class);
+        when(indexShard.getTranslog()).thenReturn(translog);
+
+        final GlobalCheckpointSyncAction action = new GlobalCheckpointSyncAction(
+            Settings.EMPTY,
+            transportService,
+            clusterService,
+            indicesService,
+            threadPool,
+            shardStateAction,
+            new ActionFilters(Collections.emptySet()),
+            new IndexNameExpressionResolver(Settings.EMPTY));
+        final ShardId shardId = new ShardId(index, id);
+        final GlobalCheckpointSyncAction.PrimaryRequest primaryRequest = new GlobalCheckpointSyncAction.PrimaryRequest(shardId);
+        if (randomBoolean()) {
+            action.shardOperationOnPrimary(primaryRequest);
+        } else {
+            action.shardOperationOnReplica(new GlobalCheckpointSyncAction.ReplicaRequest(primaryRequest, randomPositiveLong()));
+        }
+
+        verify(translog).sync();
+    }
+
+}