diff --git a/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java b/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java index fb8815741e504..ebb911c739eb3 100644 --- a/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java @@ -1071,9 +1071,6 @@ public void testCloseIndexWithNoOpSyncAndFlushForAsyncTranslog() throws Interrup }).start(); // Wait for atleast one doc to be ingested. latch.await(); - // Sleep for some time for the next doc to be present in lucene buffer. If flush happens first before the doc #2 - // gets indexed, then it goes into the happy case where the close index happens succefully. - Thread.sleep(1000); // Flush so that the subsequent sync or flushes are no-op. flush(INDEX_NAME); // Closing the index involves translog.sync and shard.flush which are now no-op. diff --git a/server/src/test/java/org/opensearch/index/translog/RemoteFsTranslogTests.java b/server/src/test/java/org/opensearch/index/translog/RemoteFsTranslogTests.java index 339d876274557..03c77a9a83f57 100644 --- a/server/src/test/java/org/opensearch/index/translog/RemoteFsTranslogTests.java +++ b/server/src/test/java/org/opensearch/index/translog/RemoteFsTranslogTests.java @@ -1801,6 +1801,83 @@ public void testDownloadWithEmptyTranslogOnlyInLocal() throws IOException { assertArrayEquals(filesPostFirstDownload, filesPostSecondDownload); } + public void testSyncWithGlobalCheckpointUpdate() throws IOException { + ArrayList ops = new ArrayList<>(); + addToTranslogAndListAndUpload(translog, ops, new Translog.Index("1", 0, primaryTerm.get(), new byte[] { 1 })); + addToTranslogAndListAndUpload(translog, ops, new Translog.Index("2", 1, primaryTerm.get(), new byte[] { 2 })); + + // Set a global checkpoint + long initialGlobalCheckpoint = 1L; + globalCheckpoint.set(initialGlobalCheckpoint); + + // Sync the translog + translog.sync(); + + // Verify that the globalCheckpointSynced is updated + assertEquals(initialGlobalCheckpoint, ((RemoteFsTranslog) translog).getLastSyncedCheckpoint().globalCheckpoint); + + // Update global checkpoint + long newGlobalCheckpoint = 2L; + globalCheckpoint.set(newGlobalCheckpoint); + + // Add a new operation and sync + addToTranslogAndListAndUpload(translog, ops, new Translog.Index("3", 2, primaryTerm.get(), new byte[] { 3 })); + translog.sync(); + + // Verify that the globalCheckpointSynced is updated to the new value + assertEquals(newGlobalCheckpoint, ((RemoteFsTranslog) translog).getLastSyncedCheckpoint().globalCheckpoint); + } + + public void testSyncNeededWithGlobalCheckpointUpdate() throws IOException { + ArrayList ops = new ArrayList<>(); + addToTranslogAndListAndUpload(translog, ops, new Translog.Index("1", 0, primaryTerm.get(), new byte[] { 1 })); + + // Set initial global checkpoint + long initialGlobalCheckpoint = 0L; + globalCheckpoint.set(initialGlobalCheckpoint); + + // Sync the translog + translog.sync(); + + // Verify that sync is not needed + assertFalse(translog.syncNeeded()); + + // Update global checkpoint + long newGlobalCheckpoint = 1L; + globalCheckpoint.set(newGlobalCheckpoint); + + // Verify that sync is now needed due to global checkpoint update + assertTrue(translog.syncNeeded()); + + // Sync again + translog.sync(); + + // Verify that sync is not needed after syncing + assertFalse(translog.syncNeeded()); + } + + public void testGlobalCheckpointUpdateDuringClose() throws IOException { + ArrayList ops = new ArrayList<>(); + addToTranslogAndListAndUpload(translog, ops, new Translog.Index("1", 0, primaryTerm.get(), new byte[] { 1 })); + + // Set initial global checkpoint + long initialGlobalCheckpoint = 0L; + globalCheckpoint.set(initialGlobalCheckpoint); + + // Sync the translog + translog.sync(); + + // Update global checkpoint + long newGlobalCheckpoint = 1L; + globalCheckpoint.set(newGlobalCheckpoint); + + // Close the translog + translog.close(); + + // Verify that the last synced checkpoint includes the updated global checkpoint + assertEquals(newGlobalCheckpoint, ((RemoteFsTranslog) translog).getLastSyncedCheckpoint().globalCheckpoint); + } + public class ThrowingBlobRepository extends FsRepository { private final Environment environment;