zhouxinyu closed pull request #220: [ROCKETMQ-332]
MappedFileQueue#findMappedFileByOffset is not thread safe, which will cause
message loss.
URL: https://github.com/apache/rocketmq/pull/220
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java
b/store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java
index 9eb3b3ab0..c30316f19 100644
--- a/store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java
+++ b/store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java
@@ -461,26 +461,39 @@ public boolean commit(final int commitLeastPages) {
*/
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;
}
- LOG_ERROR.warn("findMappedFileByOffset failure. ", e);
+
+ for (MappedFile tmpMappedFile : this.mappedFiles) {
+ if (offset >= tmpMappedFile.getFileFromOffset()
+ && offset < tmpMappedFile.getFileFromOffset() +
this.mappedFileSize) {
+ return tmpMappedFile;
+ }
+ }
+ }
+
+ if (returnFirstOnNotFound) {
+ return firstMappedFile;
}
}
} catch (Exception e) {
diff --git
a/store/src/test/java/org/apache/rocketmq/store/MappedFileQueueTest.java
b/store/src/test/java/org/apache/rocketmq/store/MappedFileQueueTest.java
index 92f1876b2..8f76051d1 100644
--- a/store/src/test/java/org/apache/rocketmq/store/MappedFileQueueTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/MappedFileQueueTest.java
@@ -229,6 +229,24 @@ public void testDeleteExpiredFileByTime() throws Exception
{
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");
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
