Add test for stale writer correctness with pinned timestamp snapshots

server/src/test/java/org/opensearch/index/remote/RemoteStoreUtilsTests.java

@@ -993,6 +993,37 @@ public void testGetPinnedTimestampLockedFilesWithPinnedTimestampsDifferentPrefix
         assertEquals(0, metadataFilePinnedTimestampCache.size());
     }
 
+    /**
+     * This test checks the case when a stale writer is uploading metadata files with higher timestamp, but lower primary
+     * term.
+     */
+    public void testGetPinnedTimestampLockedFilesForDivergentWrites() {
+        setupRemotePinnedTimestampFeature(true);
+
+        Map<Long, String> metadataFilePinnedTimestampCache = new HashMap<>();
+
+        // Pinned timestamp 7000
+        // Primary Term - Timestamp in md file
+        // 6 - 7002
+        // 3 - 6999
+        // 4 - 6998
+        // 5 - 6995
+        // 5 - 6990
+        Tuple<Map<Long, String>, Set<String>> metadataAndLocks = testGetPinnedTimestampLockedFilesWithPinnedTimestamps(
+            Map.of(7002L, 6L, 6999L, 3L, 6998L, 4L, 6995L, 5L, 6990L, 5L),
+            Set.of(4000L, 5000L, 6000L, 7000L),
+            metadataFilePinnedTimestampCache
+        );
+        Map<Long, String> metadataFiles = metadataAndLocks.v1();
+        Set<String> implicitLockedFiles = metadataAndLocks.v2();
+
+        assertEquals(1, implicitLockedFiles.size());
+        assertTrue(implicitLockedFiles.contains(metadataFiles.get(6995L)));
+        // Now we cache all the matches except the last one.
+        assertEquals(1, metadataFilePinnedTimestampCache.size());
+        assertEquals(metadataFiles.get(6995L), metadataFilePinnedTimestampCache.get(7000L));
+    }
+
     public void testFilterOutMetadataFilesBasedOnAgeFeatureDisabled() {
         setupRemotePinnedTimestampFeature(false);
         List<String> metadataFiles = new ArrayList<>();