Put back lastSyncedGlobalCheckpoint in deletion policy

The PR #27837 unintentionally changed to an in memory global checkpoint.

Author: dnhatn

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

@@ -184,7 +184,7 @@ public InternalEngine(EngineConfig engineConfig) {
                 assert translog.getGeneration() != null;
                 this.translog = translog;
                 this.snapshotDeletionPolicy = new SnapshotDeletionPolicy(
-                    new CombinedDeletionPolicy(openMode, translogDeletionPolicy, engineConfig.getGlobalCheckpointSupplier())
+                    new CombinedDeletionPolicy(openMode, translogDeletionPolicy, translog::getLastSyncedGlobalCheckpoint)
                 );
                 writer = createWriter(openMode == EngineConfig.OpenMode.CREATE_INDEX_AND_TRANSLOG);
                 updateMaxUnsafeAutoIdTimestampFromWriter(writer);

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

@@ -4242,8 +4242,14 @@ protected void commitIndexWriter(IndexWriter writer, Translog translog, String s
                     final List<IndexCommit> commits = DirectoryReader.listCommits(store.directory());
                     // Keep only one safe commit as the oldest commit.
                     final IndexCommit safeCommit = commits.get(0);
-                    assertThat(Long.parseLong(safeCommit.getUserData().get(SequenceNumbers.MAX_SEQ_NO)),
-                        lessThanOrEqualTo(lastSyncedGlobalCheckpoint));
+                    if (lastSyncedGlobalCheckpoint == SequenceNumbers.UNASSIGNED_SEQ_NO) {
+                        // If the global checkpoint is still unassigned, we keep an empty(eg. initial) commit as a safe commit.
+                        assertThat(Long.parseLong(safeCommit.getUserData().get(SequenceNumbers.MAX_SEQ_NO)),
+                            equalTo(SequenceNumbers.NO_OPS_PERFORMED));
+                    } else {
+                        assertThat(Long.parseLong(safeCommit.getUserData().get(SequenceNumbers.MAX_SEQ_NO)),
+                            lessThanOrEqualTo(lastSyncedGlobalCheckpoint));
+                    }
                     for (int i = 1; i < commits.size(); i++) {
                         assertThat(Long.parseLong(commits.get(i).getUserData().get(SequenceNumbers.MAX_SEQ_NO)),
                             greaterThan(lastSyncedGlobalCheckpoint));