You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by st...@apache.org on 2017/01/22 12:12:28 UTC
[23/31] incubator-rocketmq git commit: [ROCKETMQ-45]Delete hanged
consume queue files
[ROCKETMQ-45]Delete hanged consume queue files
Project: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/commit/11ff542c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/tree/11ff542c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/diff/11ff542c
Branch: refs/heads/ROCKETMQ-54
Commit: 11ff542c4382faf3bb7a74a80329f1fcab6a07a6
Parents: d9c398f
Author: Zhanhui Li <li...@apache.org>
Authored: Thu Jan 12 15:02:26 2017 +0800
Committer: Zhanhui Li <li...@apache.org>
Committed: Sun Jan 22 14:48:16 2017 +0800
----------------------------------------------------------------------
.../org/apache/rocketmq/store/ConsumeQueue.java | 28 +++++------
.../rocketmq/store/DefaultMessageStore.java | 14 +++---
.../org/apache/rocketmq/store/MappedFile.java | 22 ++++-----
.../apache/rocketmq/store/MappedFileQueue.java | 4 ++
.../store/schedule/ScheduleMessageService.java | 2 +-
.../rocketmq/store/MappedFileQueueTest.java | 50 ++++++++++++++++++--
6 files changed, 84 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/11ff542c/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
----------------------------------------------------------------------
diff --git a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
index c5e38db..dc1c96c 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
@@ -77,19 +77,19 @@ public class ConsumeQueue {
if (index < 0)
index = 0;
- int mapedFileSizeLogics = this.mappedFileSize;
+ int mappedFileSizeLogics = this.mappedFileSize;
MappedFile mappedFile = mappedFiles.get(index);
ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();
long processOffset = mappedFile.getFileFromOffset();
- long mapedFileOffset = 0;
+ long mappedFileOffset = 0;
while (true) {
- for (int i = 0; i < mapedFileSizeLogics; i += CQ_STORE_UNIT_SIZE) {
+ for (int i = 0; i < mappedFileSizeLogics; i += CQ_STORE_UNIT_SIZE) {
long offset = byteBuffer.getLong();
int size = byteBuffer.getInt();
long tagsCode = byteBuffer.getLong();
if (offset >= 0 && size > 0) {
- mapedFileOffset = i + CQ_STORE_UNIT_SIZE;
+ mappedFileOffset = i + CQ_STORE_UNIT_SIZE;
this.maxPhysicOffset = offset;
} else {
log.info("recover current consume queue file over, " + mappedFile.getFileName() + " "
@@ -98,7 +98,7 @@ public class ConsumeQueue {
}
}
- if (mapedFileOffset == mapedFileSizeLogics) {
+ if (mappedFileOffset == mappedFileSizeLogics) {
index++;
if (index >= mappedFiles.size()) {
@@ -109,17 +109,17 @@ public class ConsumeQueue {
mappedFile = mappedFiles.get(index);
byteBuffer = mappedFile.sliceByteBuffer();
processOffset = mappedFile.getFileFromOffset();
- mapedFileOffset = 0;
+ mappedFileOffset = 0;
log.info("recover next consume queue file, " + mappedFile.getFileName());
}
} else {
log.info("recover current consume queue queue over " + mappedFile.getFileName() + " "
- + (processOffset + mapedFileOffset));
+ + (processOffset + mappedFileOffset));
break;
}
}
- processOffset += mapedFileOffset;
+ processOffset += mappedFileOffset;
this.mappedFileQueue.setFlushedWhere(processOffset);
this.mappedFileQueue.setCommittedWhere(processOffset);
this.mappedFileQueue.truncateDirtyFiles(processOffset);
@@ -310,7 +310,7 @@ public class ConsumeQueue {
if (offsetPy >= phyMinOffset) {
this.minLogicOffset = result.getMappedFile().getFileFromOffset() + i;
- log.info("compute logics min offset: " + this.getMinOffsetInQuque() + ", topic: "
+ log.info("compute logics min offset: " + this.getMinOffsetInQueue() + ", topic: "
+ this.topic + ", queueId: " + this.queueId);
break;
}
@@ -324,7 +324,7 @@ public class ConsumeQueue {
}
}
- public long getMinOffsetInQuque() {
+ public long getMinOffsetInQueue() {
return this.minLogicOffset / CQ_STORE_UNIT_SIZE;
}
@@ -435,8 +435,8 @@ public class ConsumeQueue {
}
public long rollNextFile(final long index) {
- int mapedFileSize = this.mappedFileSize;
- int totalUnitsInFile = mapedFileSize / CQ_STORE_UNIT_SIZE;
+ int mappedFileSize = this.mappedFileSize;
+ int totalUnitsInFile = mappedFileSize / CQ_STORE_UNIT_SIZE;
return index + totalUnitsInFile - index % totalUnitsInFile;
}
@@ -463,10 +463,10 @@ public class ConsumeQueue {
}
public long getMessageTotalInQueue() {
- return this.getMaxOffsetInQuque() - this.getMinOffsetInQuque();
+ return this.getMaxOffsetInQueue() - this.getMinOffsetInQueue();
}
- public long getMaxOffsetInQuque() {
+ public long getMaxOffsetInQueue() {
return this.mappedFileQueue.getMaxOffset() / CQ_STORE_UNIT_SIZE;
}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/11ff542c/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
----------------------------------------------------------------------
diff --git a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
index 3a43c21..2594ef3 100644
--- a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+++ b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
@@ -376,8 +376,8 @@ public class DefaultMessageStore implements MessageStore {
ConsumeQueue consumeQueue = findConsumeQueue(topic, queueId);
if (consumeQueue != null) {
- minOffset = consumeQueue.getMinOffsetInQuque();
- maxOffset = consumeQueue.getMaxOffsetInQuque();
+ minOffset = consumeQueue.getMinOffsetInQueue();
+ maxOffset = consumeQueue.getMaxOffsetInQueue();
if (maxOffset == 0) {
status = GetMessageStatus.NO_MESSAGE_IN_QUEUE;
@@ -499,7 +499,7 @@ public class DefaultMessageStore implements MessageStore {
public long getMaxOffsetInQuque(String topic, int queueId) {
ConsumeQueue logic = this.findConsumeQueue(topic, queueId);
if (logic != null) {
- long offset = logic.getMaxOffsetInQuque();
+ long offset = logic.getMaxOffsetInQueue();
return offset;
}
@@ -512,7 +512,7 @@ public class DefaultMessageStore implements MessageStore {
public long getMinOffsetInQuque(String topic, int queueId) {
ConsumeQueue logic = this.findConsumeQueue(topic, queueId);
if (logic != null) {
- return logic.getMinOffsetInQuque();
+ return logic.getMinOffsetInQueue();
}
return -1;
@@ -878,8 +878,8 @@ public class DefaultMessageStore implements MessageStore {
ConsumeQueue consumeQueue = findConsumeQueue(topic, queueId);
if (consumeQueue != null) {
- minOffset = Math.max(minOffset, consumeQueue.getMinOffsetInQuque());
- maxOffset = Math.min(maxOffset, consumeQueue.getMaxOffsetInQuque());
+ minOffset = Math.max(minOffset, consumeQueue.getMinOffsetInQueue());
+ maxOffset = Math.min(maxOffset, consumeQueue.getMaxOffsetInQueue());
if (maxOffset == 0) {
return messageIds;
@@ -1220,7 +1220,7 @@ public class DefaultMessageStore implements MessageStore {
for (ConcurrentHashMap<Integer, ConsumeQueue> maps : this.consumeQueueTable.values()) {
for (ConsumeQueue logic : maps.values()) {
String key = logic.getTopic() + "-" + logic.getQueueId();
- table.put(key, logic.getMaxOffsetInQuque());
+ table.put(key, logic.getMaxOffsetInQueue());
logic.correctMinOffset(minPhyOffset);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/11ff542c/store/src/main/java/org/apache/rocketmq/store/MappedFile.java
----------------------------------------------------------------------
diff --git a/store/src/main/java/org/apache/rocketmq/store/MappedFile.java b/store/src/main/java/org/apache/rocketmq/store/MappedFile.java
index cd4b57b..feb505d 100644
--- a/store/src/main/java/org/apache/rocketmq/store/MappedFile.java
+++ b/store/src/main/java/org/apache/rocketmq/store/MappedFile.java
@@ -43,9 +43,9 @@ public class MappedFile extends ReferenceResource {
public static final int OS_PAGE_SIZE = 1024 * 4;
protected static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
- private static final AtomicLong TOTAL_MAPED_VITUAL_MEMORY = new AtomicLong(0);
+ private static final AtomicLong TOTAL_MAPPED_VIRTUAL_MEMORY = new AtomicLong(0);
- private static final AtomicInteger TOTAL_MAPED_FILES = new AtomicInteger(0);
+ private static final AtomicInteger TOTAL_MAPPED_FILES = new AtomicInteger(0);
protected final AtomicInteger wrotePosition = new AtomicInteger(0);
//ADD BY ChenYang
protected final AtomicInteger committedPosition = new AtomicInteger(0);
@@ -132,12 +132,12 @@ public class MappedFile extends ReferenceResource {
return viewed(viewedBuffer);
}
- public static int getTotalmapedfiles() {
- return TOTAL_MAPED_FILES.get();
+ public static int getTotalMappedFiles() {
+ return TOTAL_MAPPED_FILES.get();
}
- public static long getTotalMapedVitualMemory() {
- return TOTAL_MAPED_VITUAL_MEMORY.get();
+ public static long getTotalMappedVirtualMemory() {
+ return TOTAL_MAPPED_VIRTUAL_MEMORY.get();
}
public void init(final String fileName, final int fileSize, final TransientStorePool transientStorePool) throws IOException {
@@ -158,8 +158,8 @@ public class MappedFile extends ReferenceResource {
try {
this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel();
this.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize);
- TOTAL_MAPED_VITUAL_MEMORY.addAndGet(fileSize);
- TOTAL_MAPED_FILES.incrementAndGet();
+ TOTAL_MAPPED_VIRTUAL_MEMORY.addAndGet(fileSize);
+ TOTAL_MAPPED_FILES.incrementAndGet();
ok = true;
} catch (FileNotFoundException e) {
log.error("create file channel " + this.fileName + " Failed. ", e);
@@ -405,8 +405,8 @@ public class MappedFile extends ReferenceResource {
}
clean(this.mappedByteBuffer);
- TOTAL_MAPED_VITUAL_MEMORY.addAndGet(this.fileSize * (-1));
- TOTAL_MAPED_FILES.decrementAndGet();
+ TOTAL_MAPPED_VIRTUAL_MEMORY.addAndGet(this.fileSize * (-1));
+ TOTAL_MAPPED_FILES.decrementAndGet();
log.info("unmap file[REF:" + currentRef + "] " + this.fileName + " OK");
return true;
}
@@ -431,7 +431,7 @@ public class MappedFile extends ReferenceResource {
return true;
} else {
- log.warn("destroy maped file[REF:" + this.getRefCount() + "] " + this.fileName
+ log.warn("destroy mapped file[REF:" + this.getRefCount() + "] " + this.fileName
+ " Failed. cleanupOver: " + this.cleanupOver);
}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/11ff542c/store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java
----------------------------------------------------------------------
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 f98b26a..5c6c62c 100644
--- a/store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java
+++ b/store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java
@@ -397,11 +397,15 @@ public class MappedFileQueue {
log.info("physic min offset " + offset + ", logics in current mappedFile max offset "
+ maxOffsetInLogicQueue + ", delete it");
}
+ } else if (!mappedFile.isAvailable()) { // Handle hanged file.
+ log.warn("Found a hanged consume queue file, attempting to delete it.");
+ destroy = true;
} else {
log.warn("this being not executed forever.");
break;
}
+ // TODO: Externalize this hardcoded value
if (destroy && mappedFile.destroy(1000 * 60)) {
files.add(mappedFile);
deleteCount++;
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/11ff542c/store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java
----------------------------------------------------------------------
diff --git a/store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java b/store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java
index 5b42082..e08a6f5 100644
--- a/store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java
+++ b/store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java
@@ -324,7 +324,7 @@ public class ScheduleMessageService extends ConfigManager {
*/
- long cqMinOffset = cq.getMinOffsetInQuque();
+ long cqMinOffset = cq.getMinOffsetInQueue();
if (offset < cqMinOffset) {
failScheduleOffset = cqMinOffset;
log.error("schedule CQ offset invalid. offset=" + offset + ", cqMinOffset="
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/11ff542c/store/src/test/java/org/apache/rocketmq/store/MappedFileQueueTest.java
----------------------------------------------------------------------
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 90fbbd3..2d6c112 100644
--- a/store/src/test/java/org/apache/rocketmq/store/MappedFileQueueTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/MappedFileQueueTest.java
@@ -20,6 +20,8 @@
*/
package org.apache.rocketmq.store;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
@@ -30,6 +32,7 @@ import org.slf4j.LoggerFactory;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
public class MappedFileQueueTest {
@@ -55,7 +58,7 @@ public class MappedFileQueueTest {
}
@Test
- public void test_getLastMapedFile() {
+ public void test_getLastMappedFile() {
final String fixedMsg = "0123456789abcdef";
logger.debug("================================================================");
@@ -79,7 +82,7 @@ public class MappedFileQueueTest {
}
@Test
- public void test_findMapedFileByOffset() {
+ public void test_findMappedFileByOffset() {
// four-byte string.
final String fixedMsg = "abcd";
@@ -179,7 +182,7 @@ public class MappedFileQueueTest {
}
@Test
- public void test_getMapedMemorySize() {
+ public void test_getMappedMemorySize() {
final String fixedMsg = "abcd";
logger.debug("================================================================");
@@ -200,4 +203,45 @@ public class MappedFileQueueTest {
mappedFileQueue.destroy();
logger.debug("MappedFileQueue.getMappedMemorySize() OK");
}
+
+
+ @Test
+ public void test_deleteExpiredFileByOffset() {
+
+ logger.debug("================================================================");
+ MappedFileQueue mappedFileQueue =
+ new MappedFileQueue("target/unit_test_store/e", 5120, null);
+
+ for (int i = 0; i < 2048; i++) {
+ MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0);
+ assertNotNull(mappedFile);
+
+ ByteBuffer byteBuffer = ByteBuffer.allocate(ConsumeQueue.CQ_STORE_UNIT_SIZE);
+ byteBuffer.putLong(i);
+ byte[] padding = new byte[12];
+ Arrays.fill(padding, (byte)'0');
+ byteBuffer.put(padding);
+ byteBuffer.flip();
+
+ boolean result = mappedFile.appendMessage(byteBuffer.array());
+
+ assertTrue(result);
+ }
+
+ MappedFile first = mappedFileQueue.getFirstMappedFile();
+ first.hold();
+
+ int count = mappedFileQueue.deleteExpiredFileByOffset(20480, ConsumeQueue.CQ_STORE_UNIT_SIZE);
+ assertEquals(0, count);
+ first.release();
+
+ count = mappedFileQueue.deleteExpiredFileByOffset(20480, ConsumeQueue.CQ_STORE_UNIT_SIZE);
+ assertTrue(count > 0);
+ first = mappedFileQueue.getFirstMappedFile();
+ assertTrue(first.getFileFromOffset() > 0);
+
+ mappedFileQueue.shutdown(1000);
+ mappedFileQueue.destroy();
+ logger.debug("MappedFileQueue.deleteExpiredFileByOffset() OK");
+ }
}