Merge pull request #220 from zhouxinyu/ROCKETMQ-332

[ROCKETMQ-332] MappedFileQueue#findMappedFileByOffset is not thread safe, which will cause message loss.

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

@@ -461,26 +461,39 @@ public class MappedFileQueue {
      */
     public MappedFile findMappedFileByOffset(final long offset, final boolean returnFirstOnNotFound) {
         try {
-            MappedFile mappedFile = this.getFirstMappedFile();
-            if (mappedFile != null) {
-                int index = (int) ((offset / this.mappedFileSize) - (mappedFile.getFileFromOffset() / this.mappedFileSize));
-                if (index < 0 || index >= this.mappedFiles.size()) {
-                    LOG_ERROR.warn("Offset for {} not matched. Request offset: {}, index: {}, " +
-                            "mappedFileSize: {}, mappedFiles count: {}",
-                        mappedFile,
+            MappedFile firstMappedFile = this.getFirstMappedFile();
+            MappedFile lastMappedFile = this.getLastMappedFile();
+            if (firstMappedFile != null && lastMappedFile != null) {
+                if (offset < firstMappedFile.getFileFromOffset() || offset >= lastMappedFile.getFileFromOffset() + this.mappedFileSize) {
+                    LOG_ERROR.warn("Offset not matched. Request offset: {}, firstOffset: {}, lastOffset: {}, mappedFileSize: {}, mappedFiles count: {}",
                         offset,
-                        index,
+                        firstMappedFile.getFileFromOffset(),
+                        lastMappedFile.getFileFromOffset() + this.mappedFileSize,
                         this.mappedFileSize,
                         this.mappedFiles.size());
+                } else {
+                    int index = (int) ((offset / this.mappedFileSize) - (firstMappedFile.getFileFromOffset() / this.mappedFileSize));
+                    MappedFile targetFile = null;
+                    try {
+                        targetFile = this.mappedFiles.get(index);
+                    } catch (Exception ignored) {
                     }
 
-                try {
-                    return this.mappedFiles.get(index);
-                } catch (Exception e) {
-                    if (returnFirstOnNotFound) {
-                        return mappedFile;
+                    if (targetFile != null && offset >= targetFile.getFileFromOffset()
+                        && offset < targetFile.getFileFromOffset() + this.mappedFileSize) {
+                        return targetFile;
+                    }
+
+                    for (MappedFile tmpMappedFile : this.mappedFiles) {
+                        if (offset >= tmpMappedFile.getFileFromOffset()
+                            && offset < tmpMappedFile.getFileFromOffset() + this.mappedFileSize) {
+                            return tmpMappedFile;
+                        }
                     }
-                    LOG_ERROR.warn("findMappedFileByOffset failure. ", e);
+                }
+
+                if (returnFirstOnNotFound) {
+                    return firstMappedFile;
                 }
             }
         } catch (Exception e) {

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

@@ -229,6 +229,24 @@ public class MappedFileQueueTest {
         assertThat(mappedFileQueue.getMappedFiles().size()).isEqualTo(45);
     }
 
+    @Test
+    public void testFindMappedFile_ByIteration() {
+        MappedFileQueue mappedFileQueue =
+            new MappedFileQueue("target/unit_test_store/g/", 1024, null);
+        for (int i =0 ; i < 3; i++) {
+            MappedFile mappedFile = mappedFileQueue.getLastMappedFile(1024 * i);
+            mappedFile.wrotePosition.set(1024);
+        }
+
+        assertThat(mappedFileQueue.findMappedFileByOffset(1028).getFileFromOffset()).isEqualTo(1024);
+
+        // Switch two MappedFiles and verify findMappedFileByOffset method
+        MappedFile tmpFile = mappedFileQueue.getMappedFiles().get(1);
+        mappedFileQueue.getMappedFiles().set(1, mappedFileQueue.getMappedFiles().get(2));
+        mappedFileQueue.getMappedFiles().set(2, tmpFile);
+        assertThat(mappedFileQueue.findMappedFileByOffset(1028).getFileFromOffset()).isEqualTo(1024);
+    }
+
     @After
     public void destory() {
         File file = new File("target/unit_test_store");