提交 d1fa8694 编写于 作者: S shtykh_roman 提交者: Willem Jiang

[ROCKETMQ-9] Errors in rocketmq-store module.

JIRA issue: https://issues.apache.org/jira/browse/ROCKETMQ-9
上级 fed09763
...@@ -213,7 +213,7 @@ public class AllocateMappedFileService extends ServiceThread { ...@@ -213,7 +213,7 @@ public class AllocateMappedFileService extends ServiceThread {
isSuccess = true; isSuccess = true;
} }
} catch (InterruptedException e) { } catch (InterruptedException e) {
log.warn(this.getServiceName() + " service has exception, maybe by shutdown"); log.warn(this.getServiceName() + " interrupted, possibly by shutdown.");
this.hasException = true; this.hasException = true;
return false; return false;
} catch (IOException e) { } catch (IOException e) {
......
...@@ -459,27 +459,35 @@ public class MappedFileQueue { ...@@ -459,27 +459,35 @@ public class MappedFileQueue {
return result; return result;
} }
/**
* Finds a mapped file by offset.
*
* @param offset Offset.
* @param returnFirstOnNotFound If the mapped file is not found, then return the first one.
* @return Mapped file or null (when not found and returnFirstOnNotFound is <code>false</code>).
*/
public MappedFile findMappedFileByOffset(final long offset, final boolean returnFirstOnNotFound) { public MappedFile findMappedFileByOffset(final long offset, final boolean returnFirstOnNotFound) {
try { try {
MappedFile mappedFile = this.getFirstMappedFile(); MappedFile mappedFile = this.getFirstMappedFile();
if (mappedFile != null) { if (mappedFile != null) {
int index = (int) ((offset / this.mappedFileSize) - (mappedFile.getFileFromOffset() / this.mappedFileSize)); int index = (int) ((offset / this.mappedFileSize) - (mappedFile.getFileFromOffset() / this.mappedFileSize));
if (index < 0 || index >= this.mappedFiles.size()) { if (index < 0 || index >= this.mappedFiles.size()) {
LOG_ERROR.warn("findMappedFileByOffset offset not matched, request Offset: {}, index: {}, mappedFileSize: {}, mappedFiles count: {}, StackTrace: {}", LOG_ERROR.warn("Offset for {} not matched. Request offset: {}, index: {}, " +
offset, "mappedFileSize: {}, mappedFiles count: {}",
index, mappedFile,
this.mappedFileSize, offset,
this.mappedFiles.size(), index,
UtilAll.currentStackTrace()); this.mappedFileSize,
this.mappedFiles.size());
} }
try { try {
return this.mappedFiles.get(index); return this.mappedFiles.get(index);
} catch (Exception e) { } catch (Exception e) {
if (returnFirstOnNotFound) { if (returnFirstOnNotFound)
return mappedFile; return mappedFile;
}
LOG_ERROR.warn("findMappedFileByOffset failure. {}", UtilAll.currentStackTrace());
} }
} }
} catch (Exception e) { } catch (Exception e) {
......
...@@ -94,7 +94,6 @@ public class IndexFile { ...@@ -94,7 +94,6 @@ public class IndexFile {
return this.indexHeader.getIndexCount() >= this.indexNum; return this.indexHeader.getIndexCount() >= this.indexNum;
} }
public boolean destroy(final long intervalForcibly) { public boolean destroy(final long intervalForcibly) {
return this.mappedFile.destroy(intervalForcibly); return this.mappedFile.destroy(intervalForcibly);
} }
...@@ -167,8 +166,8 @@ public class IndexFile { ...@@ -167,8 +166,8 @@ public class IndexFile {
} }
} }
} else { } else {
log.warn("putKey index count " + this.indexHeader.getIndexCount() + " index max num " log.warn("Over index file capacity: index count = " + this.indexHeader.getIndexCount()
+ this.indexNum); + "; index max num = " + this.indexNum);
} }
return false; return false;
......
...@@ -49,6 +49,9 @@ public class IndexService { ...@@ -49,6 +49,9 @@ public class IndexService {
private final ArrayList<IndexFile> indexFileList = new ArrayList<IndexFile>(); private final ArrayList<IndexFile> indexFileList = new ArrayList<IndexFile>();
private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock(); private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
/** Maximum times to attempt index file creation. */
private static final int MAX_TRY_IDX_CREATE = 3;
public IndexService(final DefaultMessageStore store) { public IndexService(final DefaultMessageStore store) {
this.defaultMessageStore = store; this.defaultMessageStore = store;
...@@ -257,44 +260,44 @@ public class IndexService { ...@@ -257,44 +260,44 @@ public class IndexService {
private IndexFile putKey(IndexFile indexFile, DispatchRequest msg, String idxKey) { private IndexFile putKey(IndexFile indexFile, DispatchRequest msg, String idxKey) {
for (boolean ok = for (boolean ok = indexFile.putKey(idxKey, msg.getCommitLogOffset(), msg.getStoreTimestamp()); !ok; ) {
indexFile.putKey(idxKey, msg.getCommitLogOffset(), log.warn("Index file [" + indexFile.getFileName() + "] is full, trying to create another one");
msg.getStoreTimestamp()); !ok; ) {
log.warn("index file full, so create another one, " + indexFile.getFileName());
indexFile = retryGetAndCreateIndexFile(); indexFile = retryGetAndCreateIndexFile();
if (null == indexFile) { if (null == indexFile) {
return null; return null;
} }
ok = ok = indexFile.putKey(idxKey, msg.getCommitLogOffset(), msg.getStoreTimestamp());
indexFile.putKey(idxKey, msg.getCommitLogOffset(),
msg.getStoreTimestamp());
} }
return indexFile; return indexFile;
} }
/**
public IndexFile retryGetAndCreateIndexFile() { * Retries to get or create index file.
*
* @return {@link IndexFile} or null on failure.
*/
private IndexFile retryGetAndCreateIndexFile() {
IndexFile indexFile = null; IndexFile indexFile = null;
for (int times = 0; null == indexFile && times < MAX_TRY_IDX_CREATE; times++) {
for (int times = 0; null == indexFile && times < 3; times++) {
indexFile = this.getAndCreateLastIndexFile(); indexFile = this.getAndCreateLastIndexFile();
if (null != indexFile) if (null != indexFile)
break; break;
try { try {
log.error("try to create index file, " + times + " times"); log.error("Tried to create index file " + times + " times");
Thread.sleep(1000); Thread.sleep(1000);
} catch (InterruptedException e) { } catch (InterruptedException e) {
e.printStackTrace(); e.printStackTrace();
} }
} }
if (null == indexFile) { if (null == indexFile) {
this.defaultMessageStore.getAccessRights().makeIndexFileError(); this.defaultMessageStore.getAccessRights().makeIndexFileError();
log.error("mark index file can not build flag"); log.error("Mark index file cannot build flag");
} }
return indexFile; return indexFile;
......
...@@ -52,6 +52,7 @@ public class MappedFileQueueTest { ...@@ -52,6 +52,7 @@ public class MappedFileQueueTest {
@Test @Test
public void test_getLastMapedFile() { public void test_getLastMapedFile() {
final String fixedMsg = "0123456789abcdef"; final String fixedMsg = "0123456789abcdef";
logger.debug("================================================================"); logger.debug("================================================================");
MappedFileQueue mappedFileQueue = MappedFileQueue mappedFileQueue =
new MappedFileQueue("target/unit_test_store/a/", 1024, null); new MappedFileQueue("target/unit_test_store/a/", 1024, null);
...@@ -59,6 +60,7 @@ public class MappedFileQueueTest { ...@@ -59,6 +60,7 @@ public class MappedFileQueueTest {
for (int i = 0; i < 1024; i++) { for (int i = 0; i < 1024; i++) {
MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0); MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0);
assertTrue(mappedFile != null); assertTrue(mappedFile != null);
boolean result = mappedFile.appendMessage(fixedMsg.getBytes()); boolean result = mappedFile.appendMessage(fixedMsg.getBytes());
if (!result) { if (!result) {
logger.debug("appendMessage " + i); logger.debug("appendMessage " + i);
...@@ -74,7 +76,9 @@ public class MappedFileQueueTest { ...@@ -74,7 +76,9 @@ public class MappedFileQueueTest {
@Test @Test
public void test_findMapedFileByOffset() { public void test_findMapedFileByOffset() {
// four-byte string.
final String fixedMsg = "abcd"; final String fixedMsg = "abcd";
logger.debug("================================================================"); logger.debug("================================================================");
MappedFileQueue mappedFileQueue = MappedFileQueue mappedFileQueue =
new MappedFileQueue("target/unit_test_store/b/", 1024, null); new MappedFileQueue("target/unit_test_store/b/", 1024, null);
...@@ -82,11 +86,13 @@ public class MappedFileQueueTest { ...@@ -82,11 +86,13 @@ public class MappedFileQueueTest {
for (int i = 0; i < 1024; i++) { for (int i = 0; i < 1024; i++) {
MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0); MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0);
assertTrue(mappedFile != null); assertTrue(mappedFile != null);
boolean result = mappedFile.appendMessage(fixedMsg.getBytes()); boolean result = mappedFile.appendMessage(fixedMsg.getBytes());
// logger.debug("appendMessage " + bytes);
assertTrue(result); assertTrue(result);
} }
assertEquals(fixedMsg.getBytes().length * 1024, mappedFileQueue.getMappedMemorySize());
MappedFile mappedFile = mappedFileQueue.findMappedFileByOffset(0); MappedFile mappedFile = mappedFileQueue.findMappedFileByOffset(0);
assertTrue(mappedFile != null); assertTrue(mappedFile != null);
assertEquals(mappedFile.getFileFromOffset(), 0); assertEquals(mappedFile.getFileFromOffset(), 0);
...@@ -110,7 +116,8 @@ public class MappedFileQueueTest { ...@@ -110,7 +116,8 @@ public class MappedFileQueueTest {
mappedFile = mappedFileQueue.findMappedFileByOffset(1024 * 2 + 100); mappedFile = mappedFileQueue.findMappedFileByOffset(1024 * 2 + 100);
assertTrue(mappedFile != null); assertTrue(mappedFile != null);
assertEquals(mappedFile.getFileFromOffset(), 1024 * 2); assertEquals(mappedFile.getFileFromOffset(), 1024 * 2);
// over mapped memory size.
mappedFile = mappedFileQueue.findMappedFileByOffset(1024 * 4); mappedFile = mappedFileQueue.findMappedFileByOffset(1024 * 4);
assertTrue(mappedFile == null); assertTrue(mappedFile == null);
...@@ -125,6 +132,7 @@ public class MappedFileQueueTest { ...@@ -125,6 +132,7 @@ public class MappedFileQueueTest {
@Test @Test
public void test_commit() { public void test_commit() {
final String fixedMsg = "0123456789abcdef"; final String fixedMsg = "0123456789abcdef";
logger.debug("================================================================"); logger.debug("================================================================");
MappedFileQueue mappedFileQueue = MappedFileQueue mappedFileQueue =
new MappedFileQueue("target/unit_test_store/c/", 1024, null); new MappedFileQueue("target/unit_test_store/c/", 1024, null);
...@@ -132,6 +140,7 @@ public class MappedFileQueueTest { ...@@ -132,6 +140,7 @@ public class MappedFileQueueTest {
for (int i = 0; i < 1024; i++) { for (int i = 0; i < 1024; i++) {
MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0); MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0);
assertTrue(mappedFile != null); assertTrue(mappedFile != null);
boolean result = mappedFile.appendMessage(fixedMsg.getBytes()); boolean result = mappedFile.appendMessage(fixedMsg.getBytes());
assertTrue(result); assertTrue(result);
} }
...@@ -168,6 +177,7 @@ public class MappedFileQueueTest { ...@@ -168,6 +177,7 @@ public class MappedFileQueueTest {
@Test @Test
public void test_getMapedMemorySize() { public void test_getMapedMemorySize() {
final String fixedMsg = "abcd"; final String fixedMsg = "abcd";
logger.debug("================================================================"); logger.debug("================================================================");
MappedFileQueue mappedFileQueue = MappedFileQueue mappedFileQueue =
new MappedFileQueue("target/unit_test_store/d/", 1024, null); new MappedFileQueue("target/unit_test_store/d/", 1024, null);
...@@ -175,14 +185,15 @@ public class MappedFileQueueTest { ...@@ -175,14 +185,15 @@ public class MappedFileQueueTest {
for (int i = 0; i < 1024; i++) { for (int i = 0; i < 1024; i++) {
MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0); MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0);
assertTrue(mappedFile != null); assertTrue(mappedFile != null);
boolean result = mappedFile.appendMessage(fixedMsg.getBytes()); boolean result = mappedFile.appendMessage(fixedMsg.getBytes());
assertTrue(result); assertTrue(result);
} }
assertEquals(fixedMsg.length() * 1024, mappedFileQueue.getMappedMemorySize()); assertEquals(fixedMsg.length() * 1024, mappedFileQueue.getMappedMemorySize());
mappedFileQueue.shutdown(1000); mappedFileQueue.shutdown(1000);
mappedFileQueue.destroy(); mappedFileQueue.destroy();
logger.debug("MappedFileQueue.getMappedMemorySize() OK"); logger.debug("MappedFileQueue.getMappedMemorySize() OK");
} }
} }
...@@ -31,17 +31,18 @@ import static org.junit.Assert.assertTrue; ...@@ -31,17 +31,18 @@ import static org.junit.Assert.assertTrue;
public class IndexFileTest { public class IndexFileTest {
private static final int hashSlotNum = 100; private static final int HASH_SLOT_NUM = 100;
private static final int indexNum = 400; private static final int INDEX_NUM = 400;
@Test @Test
public void test_put_index() throws Exception { public void test_put_index() throws Exception {
IndexFile indexFile = new IndexFile("100", hashSlotNum, indexNum, 0, 0); IndexFile indexFile = new IndexFile("100", HASH_SLOT_NUM, INDEX_NUM, 0, 0);
for (long i = 0; i < (indexNum - 1); i++) { for (long i = 0; i < (INDEX_NUM - 1); i++) {
boolean putResult = indexFile.putKey(Long.toString(i), i, System.currentTimeMillis()); boolean putResult = indexFile.putKey(Long.toString(i), i, System.currentTimeMillis());
assertTrue(putResult); assertTrue(putResult);
} }
// put over index file capacity.
boolean putResult = indexFile.putKey(Long.toString(400), 400, System.currentTimeMillis()); boolean putResult = indexFile.putKey(Long.toString(400), 400, System.currentTimeMillis());
assertFalse(putResult); assertFalse(putResult);
...@@ -51,12 +52,14 @@ public class IndexFileTest { ...@@ -51,12 +52,14 @@ public class IndexFileTest {
@Test @Test
public void test_put_get_index() throws Exception { public void test_put_get_index() throws Exception {
IndexFile indexFile = new IndexFile("200", hashSlotNum, indexNum, 0, 0); IndexFile indexFile = new IndexFile("200", HASH_SLOT_NUM, INDEX_NUM, 0, 0);
for (long i = 0; i < (indexNum - 1); i++) { for (long i = 0; i < (INDEX_NUM - 1); i++) {
boolean putResult = indexFile.putKey(Long.toString(i), i, System.currentTimeMillis()); boolean putResult = indexFile.putKey(Long.toString(i), i, System.currentTimeMillis());
assertTrue(putResult); assertTrue(putResult);
} }
// put over index file capacity.
boolean putResult = indexFile.putKey(Long.toString(400), 400, System.currentTimeMillis()); boolean putResult = indexFile.putKey(Long.toString(400), 400, System.currentTimeMillis());
assertFalse(putResult); assertFalse(putResult);
...@@ -64,6 +67,7 @@ public class IndexFileTest { ...@@ -64,6 +67,7 @@ public class IndexFileTest {
indexFile.selectPhyOffset(phyOffsets, "60", 10, 0, Long.MAX_VALUE, true); indexFile.selectPhyOffset(phyOffsets, "60", 10, 0, Long.MAX_VALUE, true);
assertFalse(phyOffsets.isEmpty()); assertFalse(phyOffsets.isEmpty());
assertEquals(1, phyOffsets.size()); assertEquals(1, phyOffsets.size());
indexFile.destroy(0); indexFile.destroy(0);
} }
} }
...@@ -27,7 +27,7 @@ ...@@ -27,7 +27,7 @@
<appender-ref ref="STDOUT" /> <appender-ref ref="STDOUT" />
</logger> </logger>
<root level="WARN"> <root level="ERROR">
<appender-ref ref="STDOUT" /> <appender-ref ref="STDOUT" />
</root> </root>
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册