Fix for failur IT's

Signed-off-by: Sandeep Kumawat <skumwt@amazon.com>

Author: Sandeep Kumawat

server/src/internalClusterTest/java/org/opensearch/indices/replication/WarmIndexRemoteStoreSegmentReplicationIT.java

@@ -15,7 +15,6 @@
 import org.junit.Before;
 
 import java.nio.file.Path;
-
 /**
  * This class runs Segment Replication Integ test suite with remote store enabled.
  */

server/src/internalClusterTest/java/org/opensearch/remotestore/SegmentReplicationUsingRemoteStoreIT.java

@@ -190,6 +190,7 @@ public synchronized void updateSegments(final SegmentInfos infos) throws IOExcep
     private void commitSegmentInfos(SegmentInfos infos) throws IOException {
         // get a reference to the previous commit files so they can be decref'd once a new commit is made.
         final Collection<String> previousCommitFiles = getLastCommittedSegmentInfos().files(true);
+        logger.info("[{}] commitSegmentInfos() [{}]", shardId, previousCommitFiles);
         store.commitSegmentInfos(infos, localCheckpointTracker.getMaxSeqNo(), localCheckpointTracker.getProcessedCheckpoint());
         this.lastCommittedSegmentInfos = store.readLastCommittedSegmentsInfo();
         // incref the latest on-disk commit.
@@ -370,7 +371,7 @@ public void flush(boolean force, boolean waitIfOngoing) throws EngineException {
         ensureOpen();
         // Skip flushing for indices with partial locality (warm indices)
         // For these indices, we don't need to commit as we will sync from the remote store on re-open
-        if (engineConfig.getIndexSettings().isStoreLocalityPartial() == false) {
+        if (engineConfig.getIndexSettings().isStoreLocalityPartial()) {
             return;
         }
         // readLock is held here to wait/block any concurrent close that acquires the writeLock.
@@ -447,7 +448,9 @@ protected final void closeNoLock(String reason, CountDownLatch closedLatch) {
                     latestSegmentInfos.changed();
                 }
                 try {
-                    commitSegmentInfos(latestSegmentInfos);
+                    if (engineConfig.getIndexSettings().isStoreLocalityPartial() == false) {
+                        commitSegmentInfos(latestSegmentInfos);
+                    }
                 } catch (IOException e) {
                     // mark the store corrupted unless we are closing as result of engine failure.
                     // in this case Engine#failShard will handle store corruption.

server/src/main/java/org/opensearch/index/engine/NRTReplicationEngine.java

@@ -767,7 +767,7 @@ public void updateShardState(
                                 assert newRouting.primary() && currentRouting.primary() == false;
                                 ReplicationTimer timer = new ReplicationTimer();
                                 timer.start();
-                                logger.debug(
+                                logger.info(
                                     "Resetting engine on promotion of shard [{}] to primary, startTime {}\n",
                                     shardId,
                                     timer.startTime()
@@ -5160,8 +5160,10 @@ public void syncSegmentsFromRemoteSegmentStore(boolean overrideLocal, final Runn
                         store.deleteQuiet(file);
                     }
                 }
-                assert Arrays.stream(store.directory().listAll()).filter(f -> f.startsWith(IndexFileNames.SEGMENTS)).findAny().isEmpty()
-                    : "There should not be any segments file in the dir";
+                if (indexSettings.isStoreLocalityPartial() == false) {
+                    assert Arrays.stream(store.directory().listAll()).filter(f -> f.startsWith(IndexFileNames.SEGMENTS)).findAny().isEmpty()
+                        : "There should not be any segments file in the dir";
+                }
                 store.commitSegmentInfos(infosSnapshot, processedLocalCheckpoint, processedLocalCheckpoint);
             }
             syncSegmentSuccess = true;

server/src/main/java/org/opensearch/index/shard/IndexShard.java

@@ -86,10 +86,7 @@ public String[] listAll() throws IOException {
         logger.trace("Composite Directory[{}]: listAll() called", this::toString);
         String[] localFiles = localDirectory.listAll();
         Set<String> allFiles = new HashSet<>(Arrays.asList(localFiles));
-        // String[] remoteFiles = getRemoteFiles();
-        // allFiles.addAll(Arrays.asList(remoteFiles));
-        logger.trace("Composite Directory[{}]: Local Directory files - {}", this::toString, () -> Arrays.toString(localFiles));
-        // logger.trace("Composite Directory[{}]: Remote Directory files - {}", this::toString, () -> Arrays.toString(remoteFiles));
+        logger.trace("listAll Composite Directory[{}]: Local Directory files - {}", this::toString, () -> Arrays.toString(localFiles));
         Set<String> nonBlockLuceneFiles = allFiles.stream()
             .filter(file -> !FileTypeUtils.isBlockFile(file))
             .collect(Collectors.toUnmodifiableSet());
@@ -113,12 +110,16 @@ public void deleteFile(String name) throws IOException {
         if (FileTypeUtils.isTempFile(name)) {
             localDirectory.deleteFile(name);
         } else if (Arrays.asList(listAll()).contains(name) == false) {
-            logger.debug("The file [{}] does not exist", name);
-            // we should not fail here as localDirectory might not contain this file.
-            // throw new NoSuchFileException("File " + name + " not found in directory");
+            logger.debug("The file [{}] does not exist in local directory", name);
+            // we should not throw exception in this case as localDirectory might not contain this file.
         } else {
-            localDirectory.deleteFile(name);
-            fileCache.remove(getFilePath(name));
+            // It is possible that filecache doesn't have the file, but localdirectory contains the file. We will delete it from the localDirectory.
+            if(fileCache.get(getFilePath(name)) == null) {
+                logger.info("The file [{}] exist in local but not part of FileCache, deleting it from local", name);
+                localDirectory.deleteFile(name);
+            } else {
+                fileCache.remove(getFilePath(name));
+            }
         }
     }
 

server/src/main/java/org/opensearch/index/store/CompositeDirectory.java

@@ -61,7 +61,7 @@ public void setup() throws IOException {
 
     public void testListAll() throws IOException {
         String[] actualFileNames = compositeDirectory.listAll();
-        String[] expectedFileNames = new String[] { "_0.cfe", "_0.cfs", "_0.si", "_1.cfe", "_2.cfe", "segments_1", "temp_file.tmp" };
+        String[] expectedFileNames = new String[] { "_1.cfe", "_2.cfe", "temp_file.tmp" };
         assertArrayEquals(expectedFileNames, actualFileNames);
     }