Fsync directory after cleanup (#28604)

After copying over the Lucene segments during peer recovery, we call cleanupAndVerify which removes all other files in the directory and which then calls getMetadata to check if the resulting files are a proper index. There are two issues with this:

- the directory is not fsynced after the deletions, so that the call to getMetadata, which lists files in the directory, can get a stale view, possibly seeing a deleted corruption marker (which leads to the exception seen in #28435)
- failing to delete a corruption marker should result in a hard failure, as the shard is otherwise unusable.

server/src/main/java/org/elasticsearch/index/store/Store.java

@@ -661,17 +661,17 @@ private static void failIfCorrupted(Directory directory, ShardId shardId) throws
     public void cleanupAndVerify(String reason, MetadataSnapshot sourceMetaData) throws IOException {
         metadataLock.writeLock().lock();
         try (Lock writeLock = directory.obtainLock(IndexWriter.WRITE_LOCK_NAME)) {
-            final StoreDirectory dir = directory;
-            for (String existingFile : dir.listAll()) {
+            for (String existingFile : directory.listAll()) {
                 if (Store.isAutogenerated(existingFile) || sourceMetaData.contains(existingFile)) {
                     continue; // don't delete snapshot file, or the checksums file (note, this is extra protection since the Store won't delete checksum)
                 }
                 try {
-                    dir.deleteFile(reason, existingFile);
+                    directory.deleteFile(reason, existingFile);
                     // FNF should not happen since we hold a write lock?
                 } catch (IOException ex) {
                     if (existingFile.startsWith(IndexFileNames.SEGMENTS)
-                            || existingFile.equals(IndexFileNames.OLD_SEGMENTS_GEN)) {
+                            || existingFile.equals(IndexFileNames.OLD_SEGMENTS_GEN)
+                            || existingFile.startsWith(CORRUPTED)) {
                         // TODO do we need to also fail this if we can't delete the pending commit file?
                         // if one of those files can't be deleted we better fail the cleanup otherwise we might leave an old commit point around?
                         throw new IllegalStateException("Can't delete " + existingFile + " - cleanup failed", ex);
@@ -680,6 +680,7 @@ public void cleanupAndVerify(String reason, MetadataSnapshot sourceMetaData) thr
                     // ignore, we don't really care, will get deleted later on
                 }
             }
+            directory.syncMetaData();
             final Store.MetadataSnapshot metadataOrEmpty = getMetadata(null);
             verifyAfterCleanup(sourceMetaData, metadataOrEmpty);
         } finally {