[ISSUE #7684] Fix iterator.remove() bug (#7682)

* bugfix: CopyOnWriteArray#listIterator do not support remove action when iterating

* add testcase

---------

Co-authored-by: mipengcheng3 <mipengcheng3@jd.com>

store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java

@@ -406,6 +406,7 @@ public class MappedFileQueue implements Swappable {
         }
 
         ListIterator<MappedFile> iterator = this.mappedFiles.listIterator(mappedFiles.size());
+        List<MappedFile> toRemoves = new ArrayList<>();
 
         while (iterator.hasPrevious()) {
             mappedFileLast = iterator.previous();
@@ -416,9 +417,14 @@ public class MappedFileQueue implements Swappable {
                 mappedFileLast.setCommittedPosition(where);
                 break;
             } else {
-                iterator.remove();
+                toRemoves.add(mappedFileLast);
             }
         }
+
+        if (!toRemoves.isEmpty()) {
+            this.mappedFiles.removeAll(toRemoves);
+        }
+
         return true;
     }
 

store/src/test/java/org/apache/rocketmq/store/MappedFileQueueTest.java

@@ -477,6 +477,21 @@ public class MappedFileQueueTest {
         TimeUnit.SECONDS.sleep(3);
     }
 
+    @Test
+    public void testReset() {
+        final String fixedMsg = "0123456789abcdef";
+        MappedFileQueue mappedFileQueue =
+                new MappedFileQueue(storePath + File.separator + "a/", 64, null);
+        for (int i = 0; i < 8; i++) {
+            MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0);
+            assertThat(mappedFile).isNotNull();
+            assertThat(mappedFile.appendMessage(fixedMsg.getBytes())).isTrue();
+        }
+        assertThat(mappedFileQueue.getMappedFiles().size()).isEqualTo(2);
+        assertThat(mappedFileQueue.resetOffset(0)).isTrue();
+        assertThat(mappedFileQueue.getMappedFiles().size()).isEqualTo(1);
+    }
+
     @After
     public void destroy() {
         File file = new File(storePath);