Fix flaky test SegmentReplicationIT.testScrollWithOngoingSegmentReplication (#7572)

This test has flaky failures for two reasons:
1. Fetches list of temporary files on disk starting with ".replication" before the replica has time to
flush received chunks to disk. Fixed by wrapping the assertion that a tmp file exists with assertBusy.
2. Asserts that the count of tmp files is exactly the same before/after a scroll request is cleared. However,
it is possible that additional tmp files have been written to disk concurrently, causing a count mismatch. Fixed
by removing the size assertion.  For the sake of this test we don't care if this is the case, as long as the tmp files
originally fetched remain after a scroll query is cleared.

Signed-off-by: Marc Handalian <handalm@amazon.com>

Author: mch2

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

@@ -1045,19 +1045,24 @@ public void testScrollWithOngoingSegmentReplication() throws Exception {
         // wait for segrep to start and copy temporary files
         waitForFileCopy.await();
 
-        // verify replica contains temporary files
-        IndexShard replicaShard = getIndexShard(replica, INDEX_NAME);
-        List<String> temporaryFiles = Arrays.stream(replicaShard.store().directory().listAll())
-            .filter(fileName -> fileName.startsWith(REPLICATION_PREFIX))
-            .collect(Collectors.toList());
-        logger.info("--> temporaryFiles {}", temporaryFiles);
-        assertTrue(temporaryFiles.size() > 0);
+        final IndexShard replicaShard = getIndexShard(replica, INDEX_NAME);
+        // Wait until replica has written a tmp file to disk.
+        List<String> temporaryFiles = new ArrayList<>();
+        assertBusy(() -> {
+            // verify replica contains temporary files
+            temporaryFiles.addAll(
+                Arrays.stream(replicaShard.store().directory().listAll())
+                    .filter(fileName -> fileName.startsWith(REPLICATION_PREFIX))
+                    .collect(Collectors.toList())
+            );
+            logger.info("--> temporaryFiles {}", temporaryFiles);
+            assertTrue(temporaryFiles.size() > 0);
+        });
 
         // Clear scroll query, this should clean up files on replica
         client(replica).prepareClearScroll().addScrollId(searchResponse.getScrollId()).get();
 
         // verify temporary files still exist
-        replicaShard = getIndexShard(replica, INDEX_NAME);
         List<String> temporaryFilesPostClear = Arrays.stream(replicaShard.store().directory().listAll())
             .filter(fileName -> fileName.startsWith(REPLICATION_PREFIX))
             .collect(Collectors.toList());
@@ -1066,7 +1071,6 @@ public void testScrollWithOngoingSegmentReplication() throws Exception {
         // Unblock segment replication
         blockFileCopy.countDown();
 
-        assertEquals(temporaryFiles.size(), temporaryFilesPostClear.size());
         assertTrue(temporaryFilesPostClear.containsAll(temporaryFiles));
 
         // wait for replica to catch up and verify doc count