opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-31 02:20:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ROCKETMQ-332] Fix concurrent bug in MappedFileQueue#findMappedFileByOffset, which may cause message loss

Browse Source
This commit is contained in:
yukon
2018-01-16 21:33:12 +08:00
parent d849e0ad98
commit 04e97de8a7
2 changed files with 43 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java
+25 -13
View File
@@ -461,26 +461,38 @@ 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));
MappedFile firstMappedFile = this.getFirstMappedFile();
if (firstMappedFile != null) {
int index = (int) ((offset / this.mappedFileSize) - (firstMappedFile.getFileFromOffset() / this.mappedFileSize));
if (index < 0 || index >= this.mappedFiles.size()) {
LOG_ERROR.warn("Offset for {} not matched. Request offset: {}, index: {}, " +
"mappedFileSize: {}, mappedFiles count: {}",
mappedFile,
LOG_ERROR.warn("Offset for {} not matched. Request offset: {}, index: {}, mappedFileSize: {}, mappedFiles count: {}",
firstMappedFile,
offset,
index,
this.mappedFileSize,
this.mappedFiles.size());
} else {
MappedFile targetFile = null;
try {
targetFile = this.mappedFiles.get(index);
} catch (Exception ignored) {
}
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;
}
}
}
try {
return this.mappedFiles.get(index);
} catch (Exception e) {
if (returnFirstOnNotFound) {
return mappedFile;
}
LOG_ERROR.warn("findMappedFileByOffset failure. ", e);
if (returnFirstOnNotFound) {
return firstMappedFile;
}
}
} catch (Exception e) {
store/src/test/java/org/apache/rocketmq/store/MappedFileQueueTest.java
+18
View File
@@ -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");