propertiesMap = null;
if (propertiesLength > 0) {
@@ -355,19 +342,19 @@ public class CommitLog {
return new DispatchRequest(totalSize, false/* success */);
}
- return new DispatchRequest(//
- topic, // 1
- queueId, // 2
- physicOffset, // 3
- totalSize, // 4
- tagsCode, // 5
- storeTimestamp, // 6
- queueOffset, // 7
- keys, // 8
- uniqKey, //9
- sysFlag, // 10
- preparedTransactionOffset, // 11
- propertiesMap // 12
+ return new DispatchRequest(
+ topic,
+ queueId,
+ physicOffset,
+ totalSize,
+ tagsCode,
+ storeTimestamp,
+ queueOffset,
+ keys,
+ uniqKey,
+ sysFlag,
+ preparedTransactionOffset,
+ propertiesMap
);
} catch (Exception e) {
}
@@ -376,24 +363,23 @@ public class CommitLog {
}
private static int calMsgLength(int bodyLength, int topicLength, int propertiesLength) {
- final int msgLen = 4 // 1 TOTALSIZE
- + 4 // 2 MAGICCODE
- + 4 // 3 BODYCRC
- + 4 // 4 QUEUEID
- + 4 // 5 FLAG
- + 8 // 6 QUEUEOFFSET
- + 8 // 7 PHYSICALOFFSET
- + 4 // 8 SYSFLAG
- + 8 // 9 BORNTIMESTAMP
- + 8 // 10 BORNHOST
- + 8 // 11 STORETIMESTAMP
- + 8 // 12 STOREHOSTADDRESS
- + 4 // 13 RECONSUMETIMES
- + 8 // 14 Prepared Transaction Offset
- + 4 + (bodyLength > 0 ? bodyLength : 0) // 14 BODY
- + 1 + topicLength // 15 TOPIC
- + 2 + (propertiesLength > 0 ? propertiesLength : 0) // 16
- // propertiesLength
+ final int msgLen = 4 //TOTALSIZE
+ + 4 //MAGICCODE
+ + 4 //BODYCRC
+ + 4 //QUEUEID
+ + 4 //FLAG
+ + 8 //QUEUEOFFSET
+ + 8 //PHYSICALOFFSET
+ + 4 //SYSFLAG
+ + 8 //BORNTIMESTAMP
+ + 8 //BORNHOST
+ + 8 //STORETIMESTAMP
+ + 8 //STOREHOSTADDRESS
+ + 4 //RECONSUMETIMES
+ + 8 //Prepared Transaction Offset
+ + 4 + (bodyLength > 0 ? bodyLength : 0) //BODY
+ + 1 + topicLength //TOPIC
+ + 2 + (propertiesLength > 0 ? propertiesLength : 0) //propertiesLength
+ 0;
return msgLen;
}
@@ -500,18 +486,18 @@ public class CommitLog {
return false;
}
- if (this.defaultMessageStore.getMessageStoreConfig().isMessageIndexEnable()//
+ if (this.defaultMessageStore.getMessageStoreConfig().isMessageIndexEnable()
&& this.defaultMessageStore.getMessageStoreConfig().isMessageIndexSafe()) {
if (storeTimestamp <= this.defaultMessageStore.getStoreCheckpoint().getMinTimestampIndex()) {
- log.info("find check timestamp, {} {}", //
- storeTimestamp, //
+ log.info("find check timestamp, {} {}",
+ storeTimestamp,
UtilAll.timeMillisToHumanString(storeTimestamp));
return true;
}
} else {
if (storeTimestamp <= this.defaultMessageStore.getStoreCheckpoint().getMinTimestamp()) {
- log.info("find check timestamp, {} {}", //
- storeTimestamp, //
+ log.info("find check timestamp, {} {}",
+ storeTimestamp,
UtilAll.timeMillisToHumanString(storeTimestamp));
return true;
}
@@ -547,7 +533,7 @@ public class CommitLog {
int queueId = msg.getQueueId();
final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
- if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE//
+ if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
|| tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
// Delay Delivery
if (msg.getDelayTimeLevel() > 0) {
@@ -677,7 +663,7 @@ public class CommitLog {
if (messageExt.isWaitStoreMsgOK()) {
// Determine whether to wait
if (service.isSlaveOK(result.getWroteOffset() + result.getWroteBytes())) {
- GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
+ GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
service.putRequest(request);
service.getWaitNotifyObject().wakeupAll();
boolean flushOK =
@@ -774,7 +760,6 @@ public class CommitLog {
putMessageLock.unlock();
}
-
if (eclipseTimeInLock > 500) {
log.warn("[NOTIFYME]putMessages in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", eclipseTimeInLock, messageExtBatch.getBody().length, result);
}
@@ -789,7 +774,6 @@ public class CommitLog {
storeStatsService.getSinglePutMessageTopicTimesTotal(messageExtBatch.getTopic()).addAndGet(result.getMsgNum());
storeStatsService.getSinglePutMessageTopicSizeTotal(messageExtBatch.getTopic()).addAndGet(result.getWroteBytes());
-
handleDiskFlush(result, putMessageResult, messageExtBatch);
handleHA(result, putMessageResult, messageExtBatch);
@@ -901,8 +885,6 @@ public class CommitLog {
return diff;
}
-
-
abstract class FlushCommitLogService extends ServiceThread {
protected static final int RETRY_TIMES_OVER = 10;
}
@@ -1046,23 +1028,19 @@ public class CommitLog {
private final CountDownLatch countDownLatch = new CountDownLatch(1);
private volatile boolean flushOK = false;
-
public GroupCommitRequest(long nextOffset) {
this.nextOffset = nextOffset;
}
-
public long getNextOffset() {
return nextOffset;
}
-
public void wakeupCustomer(final boolean flushOK) {
this.flushOK = flushOK;
this.countDownLatch.countDown();
}
-
public boolean waitForFlush(long timeout) {
try {
this.countDownLatch.await(timeout, TimeUnit.MILLISECONDS);
@@ -1270,8 +1248,6 @@ public class CommitLog {
// 2 MAGICCODE
this.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE);
// 3 The remaining space may be any value
- //
-
// Here the length of the specially set maxBlank
final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
byteBuffer.put(this.msgStoreItemMemory.array(), 0, maxBlank);
@@ -1391,7 +1367,6 @@ public class CommitLog {
// 2 MAGICCODE
this.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE);
// 3 The remaining space may be any value
- //
//ignore previous read
messagesByteBuff.reset();
// Here the length of the specially set maxBlank
@@ -1449,7 +1424,6 @@ public class CommitLog {
this.maxMessageSize = size;
}
-
public ByteBuffer encode(final MessageExtBatch messageExtBatch) {
msgBatchMemory.clear(); //not thread-safe
int totalMsgLen = 0;
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 275334c05f3e45df9e628d21e08bf9865242cf2e..379162d7ddb3db1ce78e6cd071fd0544790cb310 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java
@@ -350,7 +350,7 @@ public class ConsumeQueue {
if (offsetPy >= phyMinOffset) {
this.minLogicOffset = result.getMappedFile().getFileFromOffset() + i;
log.info("Compute logical min offset: {}, topic: {}, queueId: {}",
- this.getMinOffsetInQueue(), this.topic, this.queueId);
+ this.getMinOffsetInQueue(), this.topic, this.queueId);
// This maybe not take effect, when not every consume queue has extend file.
if (isExtAddr(tagsCode)) {
minExtAddr = tagsCode;
@@ -567,9 +567,6 @@ public class ConsumeQueue {
/**
* Check {@code tagsCode} is address of extend file or tags code.
- *
- * @param tagsCode
- * @return
*/
public boolean isExtAddr(long tagsCode) {
return isExtReadEnable() && this.consumeQueueExt.isExtAddr(tagsCode);
diff --git a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueueExt.java b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueueExt.java
index 1a177e99d57abf0b4a5b49d635f1f04996a290b5..a118cde73b7e946302161a4c415347c54b4ebcbe 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueueExt.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueueExt.java
@@ -58,17 +58,17 @@ public class ConsumeQueueExt {
/**
* Constructor.
*
- * @param topic topic
- * @param queueId id of queue
- * @param storePath root dir of files to store.
+ * @param topic topic
+ * @param queueId id of queue
+ * @param storePath root dir of files to store.
* @param mappedFileSize file size
- * @param bitMapLength bit map length.
+ * @param bitMapLength bit map length.
*/
public ConsumeQueueExt(final String topic,
- final int queueId,
- final String storePath,
- final int mappedFileSize,
- final int bitMapLength) {
+ final int queueId,
+ final String storePath,
+ final int mappedFileSize,
+ final int bitMapLength) {
this.storePath = storePath;
this.mappedFileSize = mappedFileSize;
@@ -94,9 +94,6 @@ public class ConsumeQueueExt {
*
* Just test {@code address} is less than 0.
*
- *
- * @param address
- * @return
*/
public boolean isExtAddr(final long address) {
return address <= MAX_ADDR;
@@ -108,9 +105,6 @@ public class ConsumeQueueExt {
* if {@code address} is less than 0, return {@code address} - {@link java.lang.Long#MIN_VALUE};
* else, just return {@code address}
*
- *
- * @param address
- * @return
*/
public long unDecorate(final long address) {
if (isExtAddr(address)) {
@@ -126,7 +120,6 @@ public class ConsumeQueueExt {
* else, just return {@code offset}
*
*
- * @param offset
* @return ext address(value is less than 0)
*/
public long decorate(final long offset) {
@@ -140,7 +133,6 @@ public class ConsumeQueueExt {
* Get data from buffer.
*
* @param address less than 0
- * @return
*/
public CqExtUnit get(final long address) {
CqExtUnit cqExtUnit = new CqExtUnit();
@@ -154,9 +146,7 @@ public class ConsumeQueueExt {
/**
* Get data from buffer, and set to {@code cqExtUnit}
*
- * @param address less than 0
- * @param cqExtUnit
- * @return
+ * @param address less than 0
*/
public boolean get(final long address, final CqExtUnit cqExtUnit) {
if (!isExtAddr(address)) {
@@ -194,7 +184,6 @@ public class ConsumeQueueExt {
* Be careful, this method is not thread safe.
*
*
- * @param cqExtUnit
* @return success: < 0: fail: >=0
*/
public long put(final CqExtUnit cqExtUnit) {
@@ -259,8 +248,6 @@ public class ConsumeQueueExt {
/**
* Load data from file when startup.
- *
- * @return
*/
public boolean load() {
boolean result = this.mappedFileQueue.load();
@@ -379,9 +366,6 @@ public class ConsumeQueueExt {
/**
* flush buffer to file.
- *
- * @param flushLeastPages
- * @return
*/
public boolean flush(final int flushLeastPages) {
return this.mappedFileQueue.flush(flushLeastPages);
@@ -400,8 +384,6 @@ public class ConsumeQueueExt {
*
* Be careful: it's an address just when invoking this method.
*
- *
- * @return
*/
public long getMaxAddress() {
MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
@@ -413,8 +395,6 @@ public class ConsumeQueueExt {
/**
* Minus address saved in file.
- *
- * @return
*/
public long getMinAddress() {
MappedFile firstFile = this.mappedFileQueue.getFirstMappedFile();
@@ -435,7 +415,8 @@ public class ConsumeQueueExt {
public static final int MAX_EXT_UNIT_SIZE = Short.MAX_VALUE;
- public CqExtUnit() {}
+ public CqExtUnit() {
+ }
public CqExtUnit(Long tagsCode, long msgStoreTime, byte[] filterBitMap) {
this.tagsCode = tagsCode == null ? 0 : tagsCode;
@@ -468,9 +449,6 @@ public class ConsumeQueueExt {
/**
* build unit from buffer from current position.
- *
- * @param buffer
- * @return
*/
private boolean read(final ByteBuffer buffer) {
if (buffer.position() + 2 > buffer.limit()) {
@@ -507,8 +485,6 @@ public class ConsumeQueueExt {
*
* if size <= 0, nothing to do.
*
- *
- * @param buffer
*/
private void readBySkip(final ByteBuffer buffer) {
ByteBuffer temp = buffer.slice();
@@ -527,9 +503,6 @@ public class ConsumeQueueExt {
* 1. @{code container} can be null, it will be created if null.
* 2. if capacity of @{code container} is less than unit size, it will be created also.
* 3. Pls be sure that size of unit is not greater than {@link #MAX_EXT_UNIT_SIZE}
- *
- * @param container
- * @return
*/
private byte[] write(final ByteBuffer container) {
this.bitMapSize = (short) (filterBitMap == null ? 0 : filterBitMap.length);
@@ -557,8 +530,6 @@ public class ConsumeQueueExt {
/**
* Calculate unit size by current data.
- *
- * @return
*/
private int calcUnitSize() {
int sizeTemp = MIN_EXT_UNIT_SIZE + (filterBitMap == null ? 0 : filterBitMap.length);
@@ -600,16 +571,23 @@ public class ConsumeQueueExt {
@Override
public boolean equals(Object o) {
- if (this == o) return true;
- if (!(o instanceof CqExtUnit)) return false;
+ if (this == o)
+ return true;
+ if (!(o instanceof CqExtUnit))
+ return false;
CqExtUnit cqExtUnit = (CqExtUnit) o;
- if (bitMapSize != cqExtUnit.bitMapSize) return false;
- if (msgStoreTime != cqExtUnit.msgStoreTime) return false;
- if (size != cqExtUnit.size) return false;
- if (tagsCode != cqExtUnit.tagsCode) return false;
- if (!Arrays.equals(filterBitMap, cqExtUnit.filterBitMap)) return false;
+ if (bitMapSize != cqExtUnit.bitMapSize)
+ return false;
+ if (msgStoreTime != cqExtUnit.msgStoreTime)
+ return false;
+ if (size != cqExtUnit.size)
+ return false;
+ if (tagsCode != cqExtUnit.tagsCode)
+ return false;
+ if (!Arrays.equals(filterBitMap, cqExtUnit.filterBitMap))
+ return false;
return true;
}
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 b5bac3f721f62512093b953fa9d5db43658f558f..95a017aee6ff0705599a18d5d9e172c6084dacb7 100644
--- a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+++ b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
@@ -218,9 +218,6 @@ public class DefaultMessageStore implements MessageStore {
this.shutdown = false;
}
- /**
-
- */
public void shutdown() {
if (!this.shutdown) {
this.shutdown = true;
@@ -392,7 +389,7 @@ public class DefaultMessageStore implements MessageStore {
long begin = this.getCommitLog().getBeginTimeInLock();
long diff = this.systemClock.now() - begin;
- if (diff < 10000000 //
+ if (diff < 10000000
&& diff > this.messageStoreConfig.getOsPageCacheBusyTimeOutMills()) {
return true;
}
@@ -413,8 +410,9 @@ public class DefaultMessageStore implements MessageStore {
return commitLog;
}
- public GetMessageResult getMessage(final String group, final String topic, final int queueId, final long offset, final int maxMsgNums,
- final MessageFilter messageFilter) {
+ public GetMessageResult getMessage(final String group, final String topic, final int queueId, final long offset,
+ final int maxMsgNums,
+ final MessageFilter messageFilter) {
if (this.shutdown) {
log.warn("message store has shutdown, so getMessage is forbidden");
return null;
@@ -578,9 +576,6 @@ public class DefaultMessageStore implements MessageStore {
return getResult;
}
- /**
-
- */
public long getMaxOffsetInQueue(String topic, int queueId) {
ConsumeQueue logic = this.findConsumeQueue(topic, queueId);
if (logic != null) {
@@ -591,9 +586,6 @@ public class DefaultMessageStore implements MessageStore {
return 0;
}
- /**
-
- */
public long getMinOffsetInQueue(String topic, int queueId) {
ConsumeQueue logic = this.findConsumeQueue(topic, queueId);
if (logic != null) {
@@ -717,22 +709,27 @@ public class DefaultMessageStore implements MessageStore {
long minLogicOffset = logicQueue.getMinLogicOffset();
SelectMappedBufferResult result = logicQueue.getIndexBuffer(minLogicOffset / ConsumeQueue.CQ_STORE_UNIT_SIZE);
- if (result != null) {
- try {
- final long phyOffset = result.getByteBuffer().getLong();
- final int size = result.getByteBuffer().getInt();
- long storeTime = this.getCommitLog().pickupStoreTimestamp(phyOffset, size);
- return storeTime;
- } catch (Exception e) {
- } finally {
- result.release();
- }
- }
+ return getStoreTime(result);
}
return -1;
}
+ private long getStoreTime(SelectMappedBufferResult result) {
+ if (result != null) {
+ try {
+ final long phyOffset = result.getByteBuffer().getLong();
+ final int size = result.getByteBuffer().getInt();
+ long storeTime = this.getCommitLog().pickupStoreTimestamp(phyOffset, size);
+ return storeTime;
+ } catch (Exception e) {
+ } finally {
+ result.release();
+ }
+ }
+ return -1;
+ }
+
@Override
public long getEarliestMessageTime() {
final long minPhyOffset = this.getMinPhyOffset();
@@ -745,17 +742,7 @@ public class DefaultMessageStore implements MessageStore {
ConsumeQueue logicQueue = this.findConsumeQueue(topic, queueId);
if (logicQueue != null) {
SelectMappedBufferResult result = logicQueue.getIndexBuffer(consumeQueueOffset);
- if (result != null) {
- try {
- final long phyOffset = result.getByteBuffer().getLong();
- final int size = result.getByteBuffer().getInt();
- long storeTime = this.getCommitLog().pickupStoreTimestamp(phyOffset, size);
- return storeTime;
- } catch (Exception ignored) {
- } finally {
- result.release();
- }
- }
+ return getStoreTime(result);
}
return -1;
@@ -895,9 +882,9 @@ public class DefaultMessageStore implements MessageStore {
ConcurrentMap queueTable = next.getValue();
for (ConsumeQueue cq : queueTable.values()) {
cq.destroy();
- log.info("cleanUnusedTopic: {} {} ConsumeQueue cleaned", //
- cq.getTopic(), //
- cq.getQueueId() //
+ log.info("cleanUnusedTopic: {} {} ConsumeQueue cleaned",
+ cq.getTopic(),
+ cq.getQueueId()
);
this.commitLog.removeQueueFromTopicQueueTable(cq.getTopic(), cq.getQueueId());
@@ -926,17 +913,17 @@ public class DefaultMessageStore implements MessageStore {
long maxCLOffsetInConsumeQueue = nextQT.getValue().getLastOffset();
if (maxCLOffsetInConsumeQueue == -1) {
- log.warn("maybe ConsumeQueue was created just now. topic={} queueId={} maxPhysicOffset={} minLogicOffset={}.", //
- nextQT.getValue().getTopic(), //
- nextQT.getValue().getQueueId(), //
- nextQT.getValue().getMaxPhysicOffset(), //
+ log.warn("maybe ConsumeQueue was created just now. topic={} queueId={} maxPhysicOffset={} minLogicOffset={}.",
+ nextQT.getValue().getTopic(),
+ nextQT.getValue().getQueueId(),
+ nextQT.getValue().getMaxPhysicOffset(),
nextQT.getValue().getMinLogicOffset());
} else if (maxCLOffsetInConsumeQueue < minCommitLogOffset) {
log.info(
- "cleanExpiredConsumerQueue: {} {} consumer queue destroyed, minCommitLogOffset: {} maxCLOffsetInConsumeQueue: {}", //
- topic, //
- nextQT.getKey(), //
- minCommitLogOffset, //
+ "cleanExpiredConsumerQueue: {} {} consumer queue destroyed, minCommitLogOffset: {} maxCLOffsetInConsumeQueue: {}",
+ topic,
+ nextQT.getKey(),
+ minCommitLogOffset,
maxCLOffsetInConsumeQueue);
DefaultMessageStore.this.commitLog.removeQueueFromTopicQueueTable(nextQT.getValue().getTopic(),
@@ -955,7 +942,8 @@ public class DefaultMessageStore implements MessageStore {
}
}
- public Map getMessageIds(final String topic, final int queueId, long minOffset, long maxOffset, SocketAddress storeHost) {
+ public Map getMessageIds(final String topic, final int queueId, long minOffset, long maxOffset,
+ SocketAddress storeHost) {
Map messageIds = new HashMap();
if (this.shutdown) {
return messageIds;
@@ -1075,11 +1063,11 @@ public class DefaultMessageStore implements MessageStore {
ConsumeQueue logic = map.get(queueId);
if (null == logic) {
- ConsumeQueue newLogic = new ConsumeQueue(//
- topic, //
- queueId, //
- StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()), //
- this.getMessageStoreConfig().getMapedFileSizeConsumeQueue(), //
+ ConsumeQueue newLogic = new ConsumeQueue(
+ topic,
+ queueId,
+ StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),
+ this.getMessageStoreConfig().getMapedFileSizeConsumeQueue(),
this);
ConsumeQueue oldLogic = map.putIfAbsent(queueId, newLogic);
if (oldLogic != null) {
@@ -1111,7 +1099,7 @@ public class DefaultMessageStore implements MessageStore {
return false;
}
- if ((messageTotal + 1) >= maxMsgNums) {
+ if (maxMsgNums <= messageTotal) {
return true;
}
@@ -1120,7 +1108,7 @@ public class DefaultMessageStore implements MessageStore {
return true;
}
- if ((messageTotal + 1) > this.messageStoreConfig.getMaxTransferCountOnMessageInDisk()) {
+ if (messageTotal > this.messageStoreConfig.getMaxTransferCountOnMessageInDisk() - 1) {
return true;
}
} else {
@@ -1128,7 +1116,7 @@ public class DefaultMessageStore implements MessageStore {
return true;
}
- if ((messageTotal + 1) > this.messageStoreConfig.getMaxTransferCountOnMessageInMemory()) {
+ if (messageTotal > this.messageStoreConfig.getMaxTransferCountOnMessageInMemory() - 1) {
return true;
}
}
@@ -1465,11 +1453,11 @@ public class DefaultMessageStore implements MessageStore {
boolean cleanAtOnce = DefaultMessageStore.this.getMessageStoreConfig().isCleanFileForciblyEnable() && this.cleanImmediately;
- log.info("begin to delete before {} hours file. timeup: {} spacefull: {} manualDeleteFileSeveralTimes: {} cleanAtOnce: {}", //
- fileReservedTime, //
- timeup, //
- spacefull, //
- manualDeleteFileSeveralTimes, //
+ log.info("begin to delete before {} hours file. timeup: {} spacefull: {} manualDeleteFileSeveralTimes: {} cleanAtOnce: {}",
+ fileReservedTime,
+ timeup,
+ spacefull,
+ manualDeleteFileSeveralTimes,
cleanAtOnce);
fileReservedTime *= 60 * 60 * 1000;
@@ -1728,7 +1716,7 @@ public class DefaultMessageStore implements MessageStore {
private void doReput() {
for (boolean doNext = true; this.isCommitLogAvailable() && doNext; ) {
- if (DefaultMessageStore.this.getMessageStoreConfig().isDuplicationEnable() //
+ if (DefaultMessageStore.this.getMessageStoreConfig().isDuplicationEnable()
&& this.reputFromOffset >= DefaultMessageStore.this.getConfirmOffset()) {
break;
}
@@ -1754,7 +1742,7 @@ public class DefaultMessageStore implements MessageStore {
dispatchRequest.getTagsCode(), dispatchRequest.getStoreTimestamp(),
dispatchRequest.getBitMap(), dispatchRequest.getPropertiesMap());
}
- // FIXED BUG By shijia
+
this.reputFromOffset += size;
readSize += size;
if (DefaultMessageStore.this.getMessageStoreConfig().getBrokerRole() == BrokerRole.SLAVE) {
diff --git a/store/src/main/java/org/apache/rocketmq/store/DispatchRequest.java b/store/src/main/java/org/apache/rocketmq/store/DispatchRequest.java
index 3d33eaf3635834aa15f9d5a5ffce578eba652711..819bb948c758d7ecfab5a982c8b5454b5907daad 100644
--- a/store/src/main/java/org/apache/rocketmq/store/DispatchRequest.java
+++ b/store/src/main/java/org/apache/rocketmq/store/DispatchRequest.java
@@ -66,23 +66,14 @@ public class DispatchRequest {
}
public DispatchRequest(int size) {
- // 1
this.topic = "";
- // 2
this.queueId = 0;
- // 3
this.commitLogOffset = 0;
- // 4
this.msgSize = size;
- // 5
this.tagsCode = 0;
- // 6
this.storeTimestamp = 0;
- // 7
this.consumeQueueOffset = 0;
- // 8
this.keys = "";
- //9
this.uniqKey = null;
this.sysFlag = 0;
this.preparedTransactionOffset = 0;
@@ -91,23 +82,14 @@ public class DispatchRequest {
}
public DispatchRequest(int size, boolean success) {
- // 1
this.topic = "";
- // 2
this.queueId = 0;
- // 3
this.commitLogOffset = 0;
- // 4
this.msgSize = size;
- // 5
this.tagsCode = 0;
- // 6
this.storeTimestamp = 0;
- // 7
this.consumeQueueOffset = 0;
- // 8
this.keys = "";
- // 9
this.uniqKey = null;
this.sysFlag = 0;
this.preparedTransactionOffset = 0;
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 42504509aeebbac0d07b5fd0e744a20b2adad1fe..492ac5f26cfafe7848c06e5d08829c922813d0f5 100644
--- a/store/src/main/java/org/apache/rocketmq/store/MappedFile.java
+++ b/store/src/main/java/org/apache/rocketmq/store/MappedFile.java
@@ -73,7 +73,8 @@ public class MappedFile extends ReferenceResource {
init(fileName, fileSize);
}
- public MappedFile(final String fileName, final int fileSize, final TransientStorePool transientStorePool) throws IOException {
+ public MappedFile(final String fileName, final int fileSize,
+ final TransientStorePool transientStorePool) throws IOException {
init(fileName, fileSize, transientStorePool);
}
@@ -142,7 +143,8 @@ public class MappedFile extends ReferenceResource {
return TOTAL_MAPPED_VIRTUAL_MEMORY.get();
}
- public void init(final String fileName, final int fileSize, final TransientStorePool transientStorePool) throws IOException {
+ public void init(final String fileName, final int fileSize,
+ final TransientStorePool transientStorePool) throws IOException {
init(fileName, fileSize);
this.writeBuffer = transientStorePool.borrowBuffer();
this.transientStorePool = transientStorePool;
@@ -209,7 +211,7 @@ public class MappedFile extends ReferenceResource {
if (messageExt instanceof MessageExtBrokerInner) {
result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBrokerInner) messageExt);
} else if (messageExt instanceof MessageExtBatch) {
- result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBatch)messageExt);
+ result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBatch) messageExt);
} else {
return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
}
@@ -217,16 +219,14 @@ public class MappedFile extends ReferenceResource {
this.storeTimestamp = result.getStoreTimestamp();
return result;
}
- log.error("MappedFile.appendMessage return null, wrotePosition: {} fileSize: {}", currentPos, this.fileSize);
+ log.error("MappedFile.appendMessage return null, wrotePosition: {} fileSize: {}", currentPos, this.fileSize);
return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
}
-
public long getFileFromOffset() {
return this.fileFromOffset;
}
-
public boolean appendMessage(final byte[] data) {
int currentPos = this.wrotePosition.get();
@@ -247,10 +247,8 @@ public class MappedFile extends ReferenceResource {
/**
* Content of data from offset to offset + length will be wrote to file.
*
- * @param data
* @param offset The offset of the subarray to be used.
* @param length The length of the subarray to be used.
- * @return
*/
public boolean appendMessage(final byte[] data, final int offset, final int length) {
int currentPos = this.wrotePosition.get();
@@ -270,7 +268,6 @@ public class MappedFile extends ReferenceResource {
}
/**
- * @param flushLeastPages
* @return The current flushed position
*/
public int flush(final int flushLeastPages) {
@@ -404,9 +401,6 @@ public class MappedFile extends ReferenceResource {
return null;
}
- /**
-
- */
public SelectMappedBufferResult selectMappedBuffer(int pos) {
int readPosition = getReadPosition();
if (pos < readPosition && pos >= 0) {
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 a8fa36485e27c298e026fb81fcabe80cb9457f86..edf4c9184608b3fef74ce08ac5e4b7149452346d 100644
--- a/store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java
+++ b/store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java
@@ -405,7 +405,6 @@ public class MappedFileQueue {
break;
}
- // TODO: Externalize this hardcoded value
if (destroy && mappedFile.destroy(1000 * 60)) {
files.add(mappedFile);
deleteCount++;
diff --git a/store/src/main/java/org/apache/rocketmq/store/MessageArrivingListener.java b/store/src/main/java/org/apache/rocketmq/store/MessageArrivingListener.java
index dee1bc75274a10044a325f8b878b8b81bc36e3bd..bae7a16168980e4e597763688760779666892f4e 100644
--- a/store/src/main/java/org/apache/rocketmq/store/MessageArrivingListener.java
+++ b/store/src/main/java/org/apache/rocketmq/store/MessageArrivingListener.java
@@ -21,5 +21,5 @@ import java.util.Map;
public interface MessageArrivingListener {
void arriving(String topic, int queueId, long logicOffset, long tagsCode,
- long msgStoreTime, byte[] filterBitMap, Map properties);
+ long msgStoreTime, byte[] filterBitMap, Map properties);
}
diff --git a/store/src/main/java/org/apache/rocketmq/store/MessageFilter.java b/store/src/main/java/org/apache/rocketmq/store/MessageFilter.java
index 6b34758b43cdce32348c79fd68ea280a67de0151..3dd0fee2907607bcca82c9fb8f6f736302367cad 100644
--- a/store/src/main/java/org/apache/rocketmq/store/MessageFilter.java
+++ b/store/src/main/java/org/apache/rocketmq/store/MessageFilter.java
@@ -24,22 +24,20 @@ public interface MessageFilter {
* match by tags code or filter bit map which is calculated when message received
* and stored in consume queue ext.
*
- * @param tagsCode tagsCode
- * @param cqExtUnit extend unit of consume queue
- * @return
+ * @param tagsCode tagsCode
+ * @param cqExtUnit extend unit of consume queue
*/
boolean isMatchedByConsumeQueue(final Long tagsCode,
- final ConsumeQueueExt.CqExtUnit cqExtUnit);
+ final ConsumeQueueExt.CqExtUnit cqExtUnit);
/**
* match by message content which are stored in commit log.
*
{@code msgBuffer} and {@code properties} are not all null.If invoked in store,
* {@code properties} is null;If invoked in {@code PullRequestHoldService}, {@code msgBuffer} is null.
*
- * @param msgBuffer message buffer in commit log, may be null if not invoked in store.
- * @param properties message properties, should decode from buffer if null by yourself.
- * @return
+ * @param msgBuffer message buffer in commit log, may be null if not invoked in store.
+ * @param properties message properties, should decode from buffer if null by yourself.
*/
boolean isMatchedByCommitLog(final ByteBuffer msgBuffer,
- final Map properties);
+ final Map properties);
}
diff --git a/store/src/main/java/org/apache/rocketmq/store/MessageStore.java b/store/src/main/java/org/apache/rocketmq/store/MessageStore.java
index 55572ce10f4624553741be9b8ed2510f110d1f5d..907dfe2093b79d86c5b3cca4cc49e02a425a8f46 100644
--- a/store/src/main/java/org/apache/rocketmq/store/MessageStore.java
+++ b/store/src/main/java/org/apache/rocketmq/store/MessageStore.java
@@ -29,12 +29,14 @@ public interface MessageStore {
/**
* Load previously stored messages.
+ *
* @return true if success; false otherwise.
*/
boolean load();
/**
* Launch this message store.
+ *
* @throws Exception if there is any error.
*/
void start() throws Exception;
@@ -51,6 +53,7 @@ public interface MessageStore {
/**
* Store a message into store.
+ *
* @param msg Message instance to store
* @return result of store operation.
*/
@@ -58,6 +61,7 @@ public interface MessageStore {
/**
* Store a batch of messages.
+ *
* @param messageExtBatch Message batch.
* @return result of storing batch messages.
*/
@@ -80,6 +84,7 @@ public interface MessageStore {
/**
* Get maximum offset of the topic queue.
+ *
* @param topic Topic name.
* @param queueId Queue ID.
* @return Maximum offset at present.
@@ -88,6 +93,7 @@ public interface MessageStore {
/**
* Get the minimum offset of the topic queue.
+ *
* @param topic Topic name.
* @param queueId Queue ID.
* @return Minimum offset at present.
@@ -96,6 +102,7 @@ public interface MessageStore {
/**
* Get the offset of the message in the commit log, which is also known as physical offset.
+ *
* @param topic Topic of the message to lookup.
* @param queueId Queue ID.
* @param consumeQueueOffset offset of consume queue.
@@ -105,6 +112,7 @@ public interface MessageStore {
/**
* Look up the physical offset of the message whose store timestamp is as specified.
+ *
* @param topic Topic of the message.
* @param queueId Queue ID.
* @param timestamp Timestamp to look up.
@@ -114,6 +122,7 @@ public interface MessageStore {
/**
* Look up the message by given commit log offset.
+ *
* @param commitLogOffset physical offset.
* @return Message whose physical offset is as specified.
*/
@@ -121,6 +130,7 @@ public interface MessageStore {
/**
* Get one message from the specified commit log offset.
+ *
* @param commitLogOffset commit log offset.
* @return wrapped result of the message.
*/
@@ -128,6 +138,7 @@ public interface MessageStore {
/**
* Get one message from the specified commit log offset.
+ *
* @param commitLogOffset commit log offset.
* @param msgSize message size.
* @return wrapped result of the message.
@@ -136,30 +147,35 @@ public interface MessageStore {
/**
* Get the running information of this store.
+ *
* @return message store running info.
*/
String getRunningDataInfo();
/**
* Message store runtime information, which should generally contains various statistical information.
+ *
* @return runtime information of the message store in format of key-value pairs.
*/
HashMap getRuntimeInfo();
/**
* Get the maximum commit log offset.
+ *
* @return maximum commit log offset.
*/
long getMaxPhyOffset();
/**
* Get the minimum commit log offset.
+ *
* @return minimum commit log offset.
*/
long getMinPhyOffset();
/**
* Get the store time of the earliest message in the given queue.
+ *
* @param topic Topic of the messages to query.
* @param queueId Queue ID to find.
* @return store time of the earliest message.
@@ -168,12 +184,14 @@ public interface MessageStore {
/**
* Get the store time of the earliest message in this store.
+ *
* @return timestamp of the earliest message in this store.
*/
long getEarliestMessageTime();
/**
* Get the store time of the message specified.
+ *
* @param topic message topic.
* @param queueId queue ID.
* @param consumeQueueOffset consume queue offset.
@@ -183,6 +201,7 @@ public interface MessageStore {
/**
* Get the total number of the messages in the specified queue.
+ *
* @param topic Topic
* @param queueId Queue ID.
* @return total number.
@@ -191,6 +210,7 @@ public interface MessageStore {
/**
* Get the raw commit log data starting from the given offset, which should used for replication purpose.
+ *
* @param offset starting offset.
* @return commit log data.
*/
@@ -198,6 +218,7 @@ public interface MessageStore {
/**
* Append data to commit log.
+ *
* @param startOffset starting offset.
* @param data data to append.
* @return true if success; false otherwise.
@@ -211,36 +232,40 @@ public interface MessageStore {
/**
* Query messages by given key.
+ *
* @param topic topic of the message.
* @param key message key.
* @param maxNum maximum number of the messages possible.
* @param begin begin timestamp.
* @param end end timestamp.
- * @return
*/
QueryMessageResult queryMessage(final String topic, final String key, final int maxNum, final long begin,
final long end);
/**
* Update HA master address.
+ *
* @param newAddr new address.
*/
void updateHaMasterAddress(final String newAddr);
/**
* Return how much the slave falls behind.
+ *
* @return number of bytes that slave falls behind.
*/
long slaveFallBehindMuch();
/**
* Return the current timestamp of the store.
+ *
* @return current time in milliseconds since 1970-01-01.
*/
long now();
/**
* Clean unused topics.
+ *
* @param topics all valid topics.
* @return number of the topics deleted.
*/
@@ -253,6 +278,7 @@ public interface MessageStore {
/**
* Check if the given message has been swapped out of the memory.
+ *
* @param topic topic.
* @param queueId queue ID.
* @param consumeOffset consume queue offset.
@@ -262,18 +288,21 @@ public interface MessageStore {
/**
* Get number of the bytes that have been stored in commit log and not yet dispatched to consume queue.
+ *
* @return number of the bytes to dispatch.
*/
long dispatchBehindBytes();
/**
* Flush the message store to persist all data.
+ *
* @return maximum offset flushed to persistent storage device.
*/
long flush();
/**
* Reset written offset.
+ *
* @param phyOffset new offset.
* @return true if success; false otherwise.
*/
@@ -281,42 +310,49 @@ public interface MessageStore {
/**
* Get confirm offset.
+ *
* @return confirm offset.
*/
long getConfirmOffset();
/**
* Set confirm offset.
+ *
* @param phyOffset confirm offset to set.
*/
void setConfirmOffset(long phyOffset);
/**
* Check if the operation system page cache is busy or not.
+ *
* @return true if the OS page cache is busy; false otherwise.
*/
boolean isOSPageCacheBusy();
/**
* Get lock time in milliseconds of the store by far.
+ *
* @return lock time in milliseconds.
*/
long lockTimeMills();
/**
* Check if the transient store pool is deficient.
+ *
* @return true if the transient store pool is running out; false otherwise.
*/
boolean isTransientStorePoolDeficient();
/**
* Get the dispatcher list.
+ *
* @return list of the dispatcher.
*/
LinkedList getDispatcherList();
/**
* Get consume queue of the topic/queue.
+ *
* @param topic Topic.
* @param queueId Queue ID.
* @return Consume queue.
diff --git a/store/src/main/java/org/apache/rocketmq/store/PutMessageLock.java b/store/src/main/java/org/apache/rocketmq/store/PutMessageLock.java
index a03e41adf7291f60de527709d3d59a7385c0c655..758f4378832173b744d1b18110b31501dfe6093c 100644
--- a/store/src/main/java/org/apache/rocketmq/store/PutMessageLock.java
+++ b/store/src/main/java/org/apache/rocketmq/store/PutMessageLock.java
@@ -21,5 +21,6 @@ package org.apache.rocketmq.store;
*/
public interface PutMessageLock {
void lock();
+
void unlock();
}
diff --git a/store/src/main/java/org/apache/rocketmq/store/PutMessageReentrantLock.java b/store/src/main/java/org/apache/rocketmq/store/PutMessageReentrantLock.java
index 9198f1c6f51703224d7fa8e36a14d7a98ad3df5c..9aa80d867217ba3183e82c7382b6f1e0b66e6ce2 100644
--- a/store/src/main/java/org/apache/rocketmq/store/PutMessageReentrantLock.java
+++ b/store/src/main/java/org/apache/rocketmq/store/PutMessageReentrantLock.java
@@ -16,7 +16,6 @@
*/
package org.apache.rocketmq.store;
-
import java.util.concurrent.locks.ReentrantLock;
/**
diff --git a/store/src/main/java/org/apache/rocketmq/store/PutMessageSpinLock.java b/store/src/main/java/org/apache/rocketmq/store/PutMessageSpinLock.java
index baa809db9e03926dda207bcad02970d0960faed2..39a32cc444c8843161f4bcbdfd6c29d723fc4c93 100644
--- a/store/src/main/java/org/apache/rocketmq/store/PutMessageSpinLock.java
+++ b/store/src/main/java/org/apache/rocketmq/store/PutMessageSpinLock.java
@@ -16,12 +16,10 @@
*/
package org.apache.rocketmq.store;
-
import java.util.concurrent.atomic.AtomicBoolean;
/**
* Spin lock Implementation to put message, suggest using this witb low race conditions
- *
*/
public class PutMessageSpinLock implements PutMessageLock {
//true: Can lock, false : in lock.
diff --git a/store/src/main/java/org/apache/rocketmq/store/RunningFlags.java b/store/src/main/java/org/apache/rocketmq/store/RunningFlags.java
index 3dcd8611ab4af86af4b2190fdb9761ac44dc5e9a..7ff11a282a3ad1f61ce0e135d4427c24c27c369d 100644
--- a/store/src/main/java/org/apache/rocketmq/store/RunningFlags.java
+++ b/store/src/main/java/org/apache/rocketmq/store/RunningFlags.java
@@ -28,7 +28,6 @@ public class RunningFlags {
private static final int DISK_FULL_BIT = 1 << 4;
-
private volatile int flagBits = 0;
public RunningFlags() {
diff --git a/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java b/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
index 19ed2113dfaef48da58fb04b9b2a40fc58648d06..02aa84a3e6b590b43901e529b810b4812144ff72 100644
--- a/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
+++ b/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
@@ -143,7 +143,6 @@ public class MessageStoreConfig {
private int transientStorePoolSize = 5;
private boolean fastFailIfNoBufferInStorePool = false;
-
public boolean isDebugLockEnable() {
return debugLockEnable;
}
@@ -605,7 +604,8 @@ public class MessageStoreConfig {
}
/**
- * Enable transient commitLog store poll only if transientStorePoolEnable is true and the FlushDiskType is ASYNC_FLUSH
+ * Enable transient commitLog store poll only if transientStorePoolEnable is true and the FlushDiskType is
+ * ASYNC_FLUSH
*
* @return true or false
*/
diff --git a/store/src/main/java/org/apache/rocketmq/store/ha/HAConnection.java b/store/src/main/java/org/apache/rocketmq/store/ha/HAConnection.java
index 3967b643133ca3a0b005785e9524b8db4bbd9243..8b9750464f12790dcd01ddf75f0c786648df190a 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ha/HAConnection.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ha/HAConnection.java
@@ -78,10 +78,6 @@ public class HAConnection {
return socketChannel;
}
- /**
-
- *
- */
class ReadSocketService extends ServiceThread {
private static final int READ_MAX_BUFFER_SIZE = 1024 * 1024;
private final Selector selector;
@@ -194,10 +190,6 @@ public class HAConnection {
}
}
- /**
-
- *
- */
class WriteSocketService extends ServiceThread {
private final Selector selector;
private final SocketChannel socketChannel;
@@ -333,9 +325,6 @@ public class HAConnection {
HAConnection.log.info(this.getServiceName() + " service end");
}
- /**
-
- */
private boolean transferData() throws Exception {
int writeSizeZeroTimes = 0;
// Write Header
diff --git a/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java b/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java
index f507b368f07b4dd913b9f9ec857bf76f60085b10..51a8a27035a3d233663b75b9499daf94fa58ba3e 100644
--- a/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java
+++ b/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java
@@ -85,9 +85,6 @@ public class HAService {
return result;
}
- /**
-
- */
public void notifyTransferSome(final long offset) {
for (long value = this.push2SlaveMaxOffset.get(); offset > value; ) {
boolean ok = this.push2SlaveMaxOffset.compareAndSet(value, offset);
@@ -182,7 +179,9 @@ public class HAService {
this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
}
- /** {@inheritDoc} */
+ /**
+ * {@inheritDoc}
+ */
@Override
public void shutdown(final boolean interrupt) {
super.shutdown(interrupt);
@@ -194,7 +193,9 @@ public class HAService {
}
}
- /** {@inheritDoc} */
+ /**
+ * {@inheritDoc}
+ */
@Override
public void run() {
log.info(this.getServiceName() + " service started");
@@ -237,7 +238,9 @@ public class HAService {
log.info(this.getServiceName() + " service end");
}
- /** {@inheritDoc} */
+ /**
+ * {@inheritDoc}
+ */
@Override
public String getServiceName() {
return AcceptSocketService.class.getSimpleName();
@@ -374,17 +377,6 @@ public class HAService {
return !this.reportOffset.hasRemaining();
}
- // private void reallocateByteBuffer() {
- // ByteBuffer bb = ByteBuffer.allocate(READ_MAX_BUFFER_SIZE);
- // int remain = this.byteBufferRead.limit() - this.dispatchPostion;
- // bb.put(this.byteBufferRead.array(), this.dispatchPostion, remain);
- // this.dispatchPostion = 0;
- // this.byteBufferRead = bb;
- // }
-
- /**
-
- */
private void reallocateByteBuffer() {
int remain = READ_MAX_BUFFER_SIZE - this.dispatchPostion;
if (remain > 0) {
@@ -426,7 +418,6 @@ public class HAService {
break;
}
} else {
- // TODO ERROR
log.info("HAClient, processReadEvent read socket < 0");
return false;
}
@@ -598,8 +589,6 @@ public class HAService {
log.info(this.getServiceName() + " service end");
}
-
- //
// private void disableWriteFlag() {
// if (this.socketChannel != null) {
// SelectionKey sk = this.socketChannel.keyFor(this.selector);
@@ -610,8 +599,6 @@ public class HAService {
// }
// }
// }
- //
- //
// private void enableWriteFlag() {
// if (this.socketChannel != null) {
// SelectionKey sk = this.socketChannel.keyFor(this.selector);
diff --git a/store/src/main/java/org/apache/rocketmq/store/index/IndexFile.java b/store/src/main/java/org/apache/rocketmq/store/index/IndexFile.java
index 54f573270872084f189f6e6e9c55725622b6e3bc..edc24764e751c3f9429f20dd91cd64e83f9247e8 100644
--- a/store/src/main/java/org/apache/rocketmq/store/index/IndexFile.java
+++ b/store/src/main/java/org/apache/rocketmq/store/index/IndexFile.java
@@ -208,9 +208,8 @@ public class IndexFile {
if (slotValue <= invalidIndex || slotValue > this.indexHeader.getIndexCount()
|| this.indexHeader.getIndexCount() <= 1) {
- // TODO NOTFOUND
} else {
- for (int nextIndexToRead = slotValue;;) {
+ for (int nextIndexToRead = slotValue; ; ) {
if (phyOffsets.size() >= maxNum) {
break;
}
diff --git a/store/src/main/java/org/apache/rocketmq/store/index/IndexHeader.java b/store/src/main/java/org/apache/rocketmq/store/index/IndexHeader.java
index 5102a216a41f51abbd8ef8a9989f0c7d8b9c40fd..44021cd58958c3fdcd6a7bcac7a775dd4d147221 100644
--- a/store/src/main/java/org/apache/rocketmq/store/index/IndexHeader.java
+++ b/store/src/main/java/org/apache/rocketmq/store/index/IndexHeader.java
@@ -20,11 +20,6 @@ import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
-/**
-
- *
- *
- */
public class IndexHeader {
public static final int INDEX_HEADER_SIZE = 40;
private static int beginTimestampIndex = 0;
diff --git a/store/src/main/java/org/apache/rocketmq/store/index/IndexService.java b/store/src/main/java/org/apache/rocketmq/store/index/IndexService.java
index c434df58e797085072cacbd22748f98e704b2b49..b565349e12f647033b95d47ca71df88c5fd4544f 100644
--- a/store/src/main/java/org/apache/rocketmq/store/index/IndexService.java
+++ b/store/src/main/java/org/apache/rocketmq/store/index/IndexService.java
@@ -35,7 +35,9 @@ import org.slf4j.LoggerFactory;
public class IndexService {
private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
- /** Maximum times to attempt index file creation. */
+ /**
+ * Maximum times to attempt index file creation.
+ */
private static final int MAX_TRY_IDX_CREATE = 3;
private final DefaultMessageStore defaultMessageStore;
private final int hashSlotNum;
@@ -140,7 +142,7 @@ public class IndexService {
public void destroy() {
try {
- this.readWriteLock.readLock().lock();
+ this.readWriteLock.writeLock().lock();
for (IndexFile f : this.indexFileList) {
f.destroy(1000 * 3);
}
@@ -148,7 +150,7 @@ public class IndexService {
} catch (Exception e) {
log.error("destroy exception", e);
} finally {
- this.readWriteLock.readLock().unlock();
+ this.readWriteLock.writeLock().unlock();
}
}
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 25640a467e4442d46f9999de75a026311a8ff497..35b8e8565ea15b02794e5d23345c1bd99e8c0128 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
@@ -336,10 +336,7 @@ public class ScheduleMessageService extends ConfigManager {
}
} // end of if (bufferCQ != null)
else {
- /*
-
- */
long cqMinOffset = cq.getMinOffsetInQueue();
if (offset < cqMinOffset) {
failScheduleOffset = cqMinOffset;
diff --git a/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStats.java b/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStats.java
index 5555b8b6ba236e3a79803c2c9c1b06c49f52ae46..a3240a4695a63daa753fd3182ac86f10ba708adc 100644
--- a/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStats.java
+++ b/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStats.java
@@ -38,9 +38,6 @@ public class BrokerStats {
this.defaultMessageStore = defaultMessageStore;
}
- /**
-
- */
public void record() {
this.msgPutTotalYesterdayMorning = this.msgPutTotalTodayMorning;
this.msgGetTotalYesterdayMorning = this.msgGetTotalTodayMorning;
diff --git a/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java b/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java
index 1515eb477131900e26ff622985675a96ce3b05c9..64f76cabacd3a8c24ae751fae35f527f4c1ebc4f 100644
--- a/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java
+++ b/store/src/main/java/org/apache/rocketmq/store/stats/BrokerStatsManager.java
@@ -123,9 +123,11 @@ public class BrokerStatsManager {
public void incTopicPutNums(final String topic) {
this.statsTable.get(TOPIC_PUT_NUMS).addValue(topic, 1, 1);
}
+
public void incTopicPutNums(final String topic, int num, int times) {
this.statsTable.get(TOPIC_PUT_NUMS).addValue(topic, num, times);
}
+
public void incTopicPutSize(final String topic, final int size) {
this.statsTable.get(TOPIC_PUT_SIZE).addValue(topic, size, 1);
}
@@ -156,9 +158,11 @@ public class BrokerStatsManager {
public void incBrokerPutNums() {
this.statsTable.get(BROKER_PUT_NUMS).getAndCreateStatsItem(this.clusterName).getValue().incrementAndGet();
}
+
public void incBrokerPutNums(final int incValue) {
this.statsTable.get(BROKER_PUT_NUMS).getAndCreateStatsItem(this.clusterName).getValue().addAndGet(incValue);
}
+
public void incBrokerGetNums(final int incValue) {
this.statsTable.get(BROKER_GET_NUMS).getAndCreateStatsItem(this.clusterName).getValue().addAndGet(incValue);
}
@@ -173,12 +177,14 @@ public class BrokerStatsManager {
return this.statsTable.get(GROUP_GET_NUMS).getStatsDataInMinute(statsKey).getTps();
}
- public void recordDiskFallBehindTime(final String group, final String topic, final int queueId, final long fallBehind) {
+ public void recordDiskFallBehindTime(final String group, final String topic, final int queueId,
+ final long fallBehind) {
final String statsKey = String.format("%d@%s@%s", queueId, topic, group);
this.momentStatsItemSetFallTime.getAndCreateStatsItem(statsKey).getValue().set(fallBehind);
}
- public void recordDiskFallBehindSize(final String group, final String topic, final int queueId, final long fallBehind) {
+ public void recordDiskFallBehindSize(final String group, final String topic, final int queueId,
+ final long fallBehind) {
final String statsKey = String.format("%d@%s@%s", queueId, topic, group);
this.momentStatsItemSetFallSize.getAndCreateStatsItem(statsKey).getValue().set(fallBehind);
}
diff --git a/store/src/test/java/org/apache/rocketmq/store/AppendCallbackTest.java b/store/src/test/java/org/apache/rocketmq/store/AppendCallbackTest.java
index fc667b6318bf4923a7ff02c612deb763a5e04cdd..7f88d36e9a7283cc494b5d83055b6cc954671c0d 100644
--- a/store/src/test/java/org/apache/rocketmq/store/AppendCallbackTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/AppendCallbackTest.java
@@ -6,18 +6,17 @@
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
*
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
*/
package org.apache.rocketmq.store;
-
import java.io.File;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
@@ -25,18 +24,20 @@ import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+
+import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageExtBatch;
import org.apache.rocketmq.store.config.MessageStoreConfig;
+import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
-
public class AppendCallbackTest {
AppendMessageCallback callback;
@@ -44,7 +45,7 @@ public class AppendCallbackTest {
CommitLog.MessageExtBatchEncoder batchEncoder = new CommitLog.MessageExtBatchEncoder(10 * 1024 * 1024);
@Before
- public void init() throws Exception{
+ public void init() throws Exception {
MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
messageStoreConfig.setMapedFileSizeCommitLog(1024 * 8);
messageStoreConfig.setMapedFileSizeConsumeQueue(1024 * 4);
@@ -58,12 +59,16 @@ public class AppendCallbackTest {
callback = commitLog.new DefaultAppendMessageCallback(1024);
}
+ @After
+ public void destroy() {
+ UtilAll.deleteFile(new File(System.getProperty("user.home") + File.separator + "unitteststore"));
+ }
@Test
- public void testAppendMessageBatchEndOfFile() throws Exception{
- List messages = new ArrayList<>();
+ public void testAppendMessageBatchEndOfFile() throws Exception {
+ List messages = new ArrayList<>();
String topic = "test-topic";
- int queue= 0;
+ int queue = 0;
for (int i = 0; i < 10; i++) {
Message msg = new Message();
msg.setBody("body".getBytes());
@@ -75,8 +80,8 @@ public class AppendCallbackTest {
messageExtBatch.setTopic(topic);
messageExtBatch.setQueueId(queue);
messageExtBatch.setBornTimestamp(System.currentTimeMillis());
- messageExtBatch.setBornHost(new InetSocketAddress("127.0.0.1",123));
- messageExtBatch.setStoreHost(new InetSocketAddress("127.0.0.1",124));
+ messageExtBatch.setBornHost(new InetSocketAddress("127.0.0.1", 123));
+ messageExtBatch.setStoreHost(new InetSocketAddress("127.0.0.1", 124));
messageExtBatch.setBody(MessageDecoder.encodeMessages(messages));
messageExtBatch.setEncodedBuff(batchEncoder.encode(messageExtBatch));
@@ -91,11 +96,12 @@ public class AppendCallbackTest {
assertTrue(result.getMsgId().length() > 0); //should have already constructed some message ids
}
+
@Test
public void testAppendMessageBatchSucc() throws Exception {
- List messages = new ArrayList<>();
+ List messages = new ArrayList<>();
String topic = "test-topic";
- int queue= 0;
+ int queue = 0;
for (int i = 0; i < 10; i++) {
Message msg = new Message();
msg.setBody("body".getBytes());
@@ -107,8 +113,8 @@ public class AppendCallbackTest {
messageExtBatch.setTopic(topic);
messageExtBatch.setQueueId(queue);
messageExtBatch.setBornTimestamp(System.currentTimeMillis());
- messageExtBatch.setBornHost(new InetSocketAddress("127.0.0.1",123));
- messageExtBatch.setStoreHost(new InetSocketAddress("127.0.0.1",124));
+ messageExtBatch.setBornHost(new InetSocketAddress("127.0.0.1", 123));
+ messageExtBatch.setStoreHost(new InetSocketAddress("127.0.0.1", 124));
messageExtBatch.setBody(MessageDecoder.encodeMessages(messages));
messageExtBatch.setEncodedBuff(batchEncoder.encode(messageExtBatch));
@@ -123,7 +129,7 @@ public class AppendCallbackTest {
assertEquals(messages.size(), allresult.getMsgNum());
Set msgIds = new HashSet<>();
- for (String msgId: allresult.getMsgId().split(",")) {
+ for (String msgId : allresult.getMsgId().split(",")) {
assertEquals(32, msgId.length());
msgIds.add(msgId);
}
diff --git a/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueExtTest.java b/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueExtTest.java
index 5dbc5848adb13928e0900515c89bd12aee009dbe..e213a025c84ecfe27c0977a547f2635d6f17f85a 100644
--- a/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueExtTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueExtTest.java
@@ -17,6 +17,8 @@
package org.apache.rocketmq.store;
+import org.apache.rocketmq.common.UtilAll;
+import org.junit.After;
import org.junit.Test;
import java.io.File;
@@ -34,7 +36,6 @@ public class ConsumeQueueExtTest {
private static final int cqExtFileSize = 10 * unitSizeWithBitMap;
private static final int unitCount = 20;
-
protected ConsumeQueueExt genExt() {
return new ConsumeQueueExt(
topic, queueId, storePath, cqExtFileSize, bitMapLength
@@ -62,24 +63,8 @@ public class ConsumeQueueExtTest {
return cqExtUnit;
}
- protected void deleteDirectory(String rootPath) {
- File file = new File(rootPath);
- deleteFile(file);
- }
-
- protected void deleteFile(File file) {
- File[] subFiles = file.listFiles();
- if (subFiles != null) {
- for (File sub : subFiles) {
- deleteFile(sub);
- }
- }
-
- file.delete();
- }
-
protected void putSth(ConsumeQueueExt consumeQueueExt, boolean getAfterPut,
- boolean unitSameSize, int unitCount) {
+ boolean unitSameSize, int unitCount) {
for (int i = 0; i < unitCount; i++) {
ConsumeQueueExt.CqExtUnit putUnit =
unitSameSize ? genUnit(true) : genUnit(i % 2 == 0);
@@ -111,7 +96,7 @@ public class ConsumeQueueExtTest {
putSth(consumeQueueExt, true, false, unitCount);
} finally {
consumeQueueExt.destroy();
- deleteDirectory(storePath);
+ UtilAll.deleteFile(new File(storePath));
}
}
@@ -139,7 +124,7 @@ public class ConsumeQueueExtTest {
}
} finally {
consumeQueueExt.destroy();
- deleteDirectory(storePath);
+ UtilAll.deleteFile(new File(storePath));
}
}
@@ -161,7 +146,7 @@ public class ConsumeQueueExtTest {
assertThat(unit).isNull();
} finally {
consumeQueueExt.destroy();
- deleteDirectory(storePath);
+ UtilAll.deleteFile(new File(storePath));
}
}
@@ -199,7 +184,7 @@ public class ConsumeQueueExtTest {
} finally {
putCqExt.destroy();
loadCqExt.destroy();
- deleteDirectory(storePath);
+ UtilAll.deleteFile(new File(storePath));
}
}
@@ -222,7 +207,7 @@ public class ConsumeQueueExtTest {
assertThat(expectMinAddress).isEqualTo(minAddress);
} finally {
consumeQueueExt.destroy();
- deleteDirectory(storePath);
+ UtilAll.deleteFile(new File(storePath));
}
}
@@ -245,7 +230,12 @@ public class ConsumeQueueExtTest {
assertThat(expectMaxAddress).isEqualTo(maxAddress);
} finally {
consumeQueueExt.destroy();
- deleteDirectory(storePath);
+ UtilAll.deleteFile(new File(storePath));
}
}
+
+ @After
+ public void destroy() {
+ UtilAll.deleteFile(new File(storePath));
+ }
}
diff --git a/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java b/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java
index 9c42fb91d24a112a871c26a2104e8b1781f26e79..b03f2fce7a0d2dd5aacc35d1bcbb3430c718bfd2 100644
--- a/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/ConsumeQueueTest.java
@@ -18,6 +18,7 @@
package org.apache.rocketmq.store;
import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
@@ -82,9 +83,8 @@ public class ConsumeQueueTest {
return msg;
}
-
public MessageStoreConfig buildStoreConfig(int commitLogFileSize, int cqFileSize,
- boolean enableCqExt, int cqExtFileSize) {
+ boolean enableCqExt, int cqExtFileSize) {
MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
messageStoreConfig.setMapedFileSizeCommitLog(commitLogFileSize);
messageStoreConfig.setMapedFileSizeConsumeQueue(cqFileSize);
@@ -111,7 +111,7 @@ public class ConsumeQueueTest {
new MessageArrivingListener() {
@Override
public void arriving(String topic, int queueId, long logicOffset, long tagsCode,
- long msgStoreTime, byte[] filterBitMap, Map properties) {
+ long msgStoreTime, byte[] filterBitMap, Map properties) {
}
}
, brokerConfig);
@@ -131,22 +131,6 @@ public class ConsumeQueueTest {
}
}
- protected void deleteDirectory(String rootPath) {
- File file = new File(rootPath);
- deleteFile(file);
- }
-
- protected void deleteFile(File file) {
- File[] subFiles = file.listFiles();
- if (subFiles != null) {
- for (File sub : subFiles) {
- deleteFile(sub);
- }
- }
-
- file.delete();
- }
-
@Test
public void testConsumeQueueWithExtendData() {
DefaultMessageStore master = null;
@@ -220,7 +204,7 @@ public class ConsumeQueueTest {
} finally {
master.shutdown();
master.destroy();
- deleteDirectory(storePath);
+ UtilAll.deleteFile(new File(storePath));
}
}
}
diff --git a/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java b/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
index 75f1de94128c0f6b329a86d9e08621f4e89088f5..6e37b705f7435003edae4eeac3654566f7a6cacc 100644
--- a/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java
@@ -17,14 +17,19 @@
package org.apache.rocketmq.store;
+import java.io.File;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
+
import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.store.config.FlushDiskType;
import org.apache.rocketmq.store.config.MessageStoreConfig;
+import org.junit.After;
+import org.apache.rocketmq.store.stats.BrokerStatsManager;
import org.junit.Before;
import org.junit.Test;
@@ -38,11 +43,37 @@ public class DefaultMessageStoreTest {
private SocketAddress BornHost;
private SocketAddress StoreHost;
private byte[] MessageBody;
+ private MessageStore messageStore;
@Before
public void init() throws Exception {
StoreHost = new InetSocketAddress(InetAddress.getLocalHost(), 8123);
BornHost = new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 0);
+
+ messageStore = buildMessageStore();
+ boolean load = messageStore.load();
+ assertTrue(load);
+ messageStore.start();
+ }
+
+ @After
+ public void destory() {
+ messageStore.shutdown();
+ messageStore.destroy();
+
+ MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
+ File file = new File(messageStoreConfig.getStorePathRootDir());
+ UtilAll.deleteFile(file);
+ }
+
+ public MessageStore buildMessageStore() throws Exception {
+ MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
+ messageStoreConfig.setMapedFileSizeCommitLog(1024 * 1024 * 10);
+ messageStoreConfig.setMapedFileSizeConsumeQueue(1024 * 1024 * 10);
+ messageStoreConfig.setMaxHashSlotNum(10000);
+ messageStoreConfig.setMaxIndexNum(100 * 100);
+ messageStoreConfig.setFlushDiskType(FlushDiskType.SYNC_FLUSH);
+ return new DefaultMessageStore(messageStoreConfig, new BrokerStatsManager("simpleTest"), new MyMessageArrivingListener(), new BrokerConfig());
}
@Test
@@ -50,32 +81,16 @@ public class DefaultMessageStoreTest {
long totalMsgs = 100;
QUEUE_TOTAL = 1;
MessageBody = StoreMessage.getBytes();
+ for (long i = 0; i < totalMsgs; i++) {
+ messageStore.putMessage(buildMessage());
+ }
- MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
- messageStoreConfig.setMapedFileSizeCommitLog(1024 * 8);
- messageStoreConfig.setMapedFileSizeConsumeQueue(1024 * 4);
- messageStoreConfig.setMaxHashSlotNum(100);
- messageStoreConfig.setMaxIndexNum(100 * 10);
- MessageStore master = new DefaultMessageStore(messageStoreConfig, null, new MyMessageArrivingListener(), new BrokerConfig());
-
- boolean load = master.load();
- assertTrue(load);
-
- master.start();
- try {
- for (long i = 0; i < totalMsgs; i++) {
- master.putMessage(buildMessage());
- }
-
- for (long i = 0; i < totalMsgs; i++) {
- GetMessageResult result = master.getMessage("GROUP_A", "TOPIC_A", 0, i, 1024 * 1024, null);
- assertThat(result).isNotNull();
- result.release();
- }
- } finally {
- master.shutdown();
- master.destroy();
+ for (long i = 0; i < totalMsgs; i++) {
+ GetMessageResult result = messageStore.getMessage("GROUP_A", "TOPIC_A", 0, i, 1024 * 1024, null);
+ assertThat(result).isNotNull();
+ result.release();
}
+ verifyThatMasterIsFunctional(totalMsgs, messageStore);
}
public MessageExtBrokerInner buildMessage() {
@@ -86,7 +101,7 @@ public class DefaultMessageStoreTest {
msg.setBody(MessageBody);
msg.setKeys(String.valueOf(System.currentTimeMillis()));
msg.setQueueId(Math.abs(QueueId.getAndIncrement()) % QUEUE_TOTAL);
- msg.setSysFlag(4);
+ msg.setSysFlag(0);
msg.setBornTimestamp(System.currentTimeMillis());
msg.setStoreHost(StoreHost);
msg.setBornHost(BornHost);
@@ -95,38 +110,61 @@ public class DefaultMessageStoreTest {
@Test
public void testGroupCommit() throws Exception {
- long totalMsgs = 100;
+ long totalMsgs = 10;
QUEUE_TOTAL = 1;
MessageBody = StoreMessage.getBytes();
- MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
- messageStoreConfig.setMapedFileSizeCommitLog(1024 * 8);
- messageStoreConfig.setFlushDiskType(FlushDiskType.SYNC_FLUSH);
- MessageStore master = new DefaultMessageStore(messageStoreConfig, null, new MyMessageArrivingListener(), new BrokerConfig());
- boolean load = master.load();
- assertTrue(load);
+ for (long i = 0; i < totalMsgs; i++) {
+ messageStore.putMessage(buildMessage());
+ }
+
+ for (long i = 0; i < totalMsgs; i++) {
+ GetMessageResult result = messageStore.getMessage("GROUP_A", "TOPIC_A", 0, i, 1024 * 1024, null);
+ assertThat(result).isNotNull();
+ result.release();
+ }
+ verifyThatMasterIsFunctional(totalMsgs, messageStore);
+ }
+
+ private void verifyThatMasterIsFunctional(long totalMsgs, MessageStore master) {
+ for (long i = 0; i < totalMsgs; i++) {
+ master.putMessage(buildMessage());
+ }
+
+ for (long i = 0; i < totalMsgs; i++) {
+ GetMessageResult result = master.getMessage("GROUP_A", "TOPIC_A", 0, i, 1024 * 1024, null);
+ assertThat(result).isNotNull();
+ result.release();
+
+ }
+ }
- master.start();
- try {
- for (long i = 0; i < totalMsgs; i++) {
- master.putMessage(buildMessage());
- }
-
- for (long i = 0; i < totalMsgs; i++) {
- GetMessageResult result = master.getMessage("GROUP_A", "TOPIC_A", 0, i, 1024 * 1024, null);
- assertThat(result).isNotNull();
- result.release();
-
- }
- } finally {
- master.shutdown();
- master.destroy();
+ @Test
+ public void testPullSize() throws Exception {
+ String topic = "pullSizeTopic";
+
+ for (int i = 0; i < 32; i++) {
+ MessageExtBrokerInner messageExtBrokerInner = buildMessage();
+ messageExtBrokerInner.setTopic(topic);
+ messageExtBrokerInner.setQueueId(0);
+ messageStore.putMessage(messageExtBrokerInner);
}
+ //wait for consume queue build
+ Thread.sleep(10);
+ String group = "simple";
+ GetMessageResult getMessageResult32 = messageStore.getMessage(group, topic, 0, 0, 32, null);
+ assertThat(getMessageResult32.getMessageBufferList().size()).isEqualTo(32);
+
+ GetMessageResult getMessageResult20 = messageStore.getMessage(group, topic, 0, 0, 20, null);
+ assertThat(getMessageResult20.getMessageBufferList().size()).isEqualTo(20);
+
+ GetMessageResult getMessageResult45 = messageStore.getMessage(group, topic, 0, 0, 10, null);
+ assertThat(getMessageResult45.getMessageBufferList().size()).isEqualTo(10);
}
private class MyMessageArrivingListener implements MessageArrivingListener {
@Override
public void arriving(String topic, int queueId, long logicOffset, long tagsCode, long msgStoreTime,
- byte[] filterBitMap, Map properties) {
+ byte[] filterBitMap, Map properties) {
}
}
}
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 f1f9c1fb1b001c45bab9f8b19f783eb07fd85b50..203dfcd565f4d5491c648f0219c99eff925f3330 100644
--- a/store/src/test/java/org/apache/rocketmq/store/MappedFileQueueTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/MappedFileQueueTest.java
@@ -17,8 +17,13 @@
package org.apache.rocketmq.store;
+import java.io.File;
import java.nio.ByteBuffer;
import java.util.Arrays;
+
+import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.store.config.MessageStoreConfig;
+import org.junit.After;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
@@ -176,4 +181,10 @@ public class MappedFileQueueTest {
mappedFileQueue.shutdown(1000);
mappedFileQueue.destroy();
}
+
+ @After
+ public void destory() {
+ File file = new File("target/unit_test_store");
+ UtilAll.deleteFile(file);
+ }
}
diff --git a/store/src/test/java/org/apache/rocketmq/store/MappedFileTest.java b/store/src/test/java/org/apache/rocketmq/store/MappedFileTest.java
index d2736acd077b2316a8beb530062ebb2c6f92ea16..50d0ae47f03a0a729f1baf9ac1206fa3db9a172d 100644
--- a/store/src/test/java/org/apache/rocketmq/store/MappedFileTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/MappedFileTest.java
@@ -20,7 +20,11 @@
*/
package org.apache.rocketmq.store;
+import java.io.File;
import java.io.IOException;
+
+import org.apache.rocketmq.common.UtilAll;
+import org.junit.After;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
@@ -47,4 +51,10 @@ public class MappedFileTest {
assertThat(mappedFile.isCleanupOver()).isTrue();
assertThat(mappedFile.destroy(1000)).isTrue();
}
+
+ @After
+ public void destory() {
+ File file = new File("target/unit_test_store");
+ UtilAll.deleteFile(file);
+ }
}
diff --git a/store/src/test/java/org/apache/rocketmq/store/StoreCheckpointTest.java b/store/src/test/java/org/apache/rocketmq/store/StoreCheckpointTest.java
index 1af63d4f27515bf52ce0b811847ac7a3898be24b..3c0d9253afa9862540aa5c0e71f99b2063e51e78 100644
--- a/store/src/test/java/org/apache/rocketmq/store/StoreCheckpointTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/StoreCheckpointTest.java
@@ -20,7 +20,11 @@
*/
package org.apache.rocketmq.store;
+import java.io.File;
import java.io.IOException;
+
+import org.apache.rocketmq.common.UtilAll;
+import org.junit.After;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
@@ -42,4 +46,10 @@ public class StoreCheckpointTest {
assertThat(storeCheckpoint.getPhysicMsgTimestamp()).isEqualTo(physicMsgTimestamp);
assertThat(storeCheckpoint.getLogicsMsgTimestamp()).isEqualTo(logicsMsgTimestamp);
}
+
+ @After
+ public void destory() {
+ File file = new File("target/checkpoint_test");
+ UtilAll.deleteFile(file);
+ }
}
diff --git a/store/src/test/java/org/apache/rocketmq/store/index/IndexFileTest.java b/store/src/test/java/org/apache/rocketmq/store/index/IndexFileTest.java
index f172e65a53827b2411300bc759ff892670024db8..7ad5b38db1d8ddb571bd2a934bfb542992c5ba91 100644
--- a/store/src/test/java/org/apache/rocketmq/store/index/IndexFileTest.java
+++ b/store/src/test/java/org/apache/rocketmq/store/index/IndexFileTest.java
@@ -20,8 +20,11 @@
*/
package org.apache.rocketmq.store.index;
+import java.io.File;
import java.util.ArrayList;
import java.util.List;
+
+import org.apache.rocketmq.common.UtilAll;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
@@ -42,6 +45,8 @@ public class IndexFileTest {
boolean putResult = indexFile.putKey(Long.toString(400), 400, System.currentTimeMillis());
assertThat(putResult).isFalse();
indexFile.destroy(0);
+ File file = new File("100");
+ UtilAll.deleteFile(file);
}
@Test
@@ -62,5 +67,7 @@ public class IndexFileTest {
assertThat(phyOffsets).isNotEmpty();
assertThat(phyOffsets.size()).isEqualTo(1);
indexFile.destroy(0);
+ File file = new File("200");
+ UtilAll.deleteFile(file);
}
}
diff --git a/style/rmq_checkstyle.xml b/style/rmq_checkstyle.xml
index 6ec2ad08fdde9efd531eba31768e20b2fb6c2d53..e93b353aaa18a00cbebd37784b52beb036fecc97 100644
--- a/style/rmq_checkstyle.xml
+++ b/style/rmq_checkstyle.xml
@@ -126,8 +126,8 @@
-
-
+
+
diff --git a/test/src/main/java/org/apache/rocketmq/test/client/rmq/RMQSqlConsumer.java b/test/src/main/java/org/apache/rocketmq/test/client/rmq/RMQSqlConsumer.java
index cb0210f44ec51fb68f71e0571efae3899130f30d..3c03ee714e68c83c89642385254e6553b0f1b097 100644
--- a/test/src/main/java/org/apache/rocketmq/test/client/rmq/RMQSqlConsumer.java
+++ b/test/src/main/java/org/apache/rocketmq/test/client/rmq/RMQSqlConsumer.java
@@ -24,6 +24,7 @@ import org.apache.rocketmq.test.listener.AbstractListener;
public class RMQSqlConsumer extends RMQNormalConsumer {
private static Logger logger = Logger.getLogger(RMQSqlConsumer.class);
private MessageSelector selector;
+
public RMQSqlConsumer(String nsAddr, String topic, MessageSelector selector,
String consumerGroup, AbstractListener listener) {
super(nsAddr, topic, "*", consumerGroup, listener);
diff --git a/test/src/test/java/org/apache/rocketmq/test/base/BaseConf.java b/test/src/test/java/org/apache/rocketmq/test/base/BaseConf.java
index 92f77b8533d34e0a3015455fc21c6b1d34f15a25..8516779e4363149ac5d206113c553297978f60d4 100644
--- a/test/src/test/java/org/apache/rocketmq/test/base/BaseConf.java
+++ b/test/src/test/java/org/apache/rocketmq/test/base/BaseConf.java
@@ -41,7 +41,7 @@ public class BaseConf {
protected static String clusterName;
protected static int brokerNum;
protected static int waitTime = 5;
- protected static int consumeTime = 1 * 60 * 1000;
+ protected static int consumeTime = 5 * 60 * 1000;
protected static NamesrvController namesrvController;
protected static BrokerController brokerController1;
protected static BrokerController brokerController2;
diff --git a/test/src/test/java/org/apache/rocketmq/test/base/IntegrationTestBase.java b/test/src/test/java/org/apache/rocketmq/test/base/IntegrationTestBase.java
index 07af7aa6e3d3ac101d0b984f51705ae45dfd912e..e1b8c9102d13490ffa9a223acf5c494c9482fc69 100644
--- a/test/src/test/java/org/apache/rocketmq/test/base/IntegrationTestBase.java
+++ b/test/src/test/java/org/apache/rocketmq/test/base/IntegrationTestBase.java
@@ -25,6 +25,7 @@ import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.namesrv.NamesrvConfig;
import org.apache.rocketmq.namesrv.NamesrvController;
import org.apache.rocketmq.remoting.netty.NettyClientConfig;
@@ -54,7 +55,8 @@ public class IntegrationTestBase {
static {
Runtime.getRuntime().addShutdownHook(new Thread() {
- @Override public void run() {
+ @Override
+ public void run() {
try {
for (BrokerController brokerController : BROKER_CONTROLLERS) {
if (brokerController != null) {
@@ -75,9 +77,9 @@ public class IntegrationTestBase {
}
}
for (File file : TMPE_FILES) {
- deleteFile(file);
+ UtilAll.deleteFile(file);
}
- } catch (Exception e){
+ } catch (Exception e) {
logger.error("Shutdown error", e);
}
}
@@ -148,12 +150,12 @@ public class IntegrationTestBase {
return brokerController;
}
- public static boolean initTopic(String topic, String nsAddr, String clusterName) {
+ public static boolean initTopic(String topic, String nsAddr, String clusterName, int queueNumbers) {
long startTime = System.currentTimeMillis();
boolean createResult;
while (true) {
- createResult = MQAdmin.createTopic(nsAddr, clusterName, topic, 8);
+ createResult = MQAdmin.createTopic(nsAddr, clusterName, topic, queueNumbers);
if (createResult) {
break;
} else if (System.currentTimeMillis() - startTime > topicCreateTime) {
@@ -169,6 +171,10 @@ public class IntegrationTestBase {
return createResult;
}
+ public static boolean initTopic(String topic, String nsAddr, String clusterName) {
+ return initTopic(topic, nsAddr, clusterName, 8);
+ }
+
public static void deleteFile(File file) {
if (!file.exists()) {
return;
diff --git a/test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/order/OrderMsgBroadCastIT.java b/test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/order/OrderMsgBroadCastIT.java
index 2c9abc067fca1a4f323cec5f7ab5b78d0e1ae049..9d8aeb3313ec20eb71a2a6b556318673134f79dd 100644
--- a/test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/order/OrderMsgBroadCastIT.java
+++ b/test/src/test/java/org/apache/rocketmq/test/client/consumer/broadcast/order/OrderMsgBroadCastIT.java
@@ -38,6 +38,8 @@ public class OrderMsgBroadCastIT extends BaseBroadCastIT {
private RMQNormalProducer producer = null;
private String topic = null;
+ private int broadcastConsumeTime = 1 * 60 * 1000;
+
@Before
public void setUp() {
topic = initTopic();
@@ -63,9 +65,8 @@ public class OrderMsgBroadCastIT extends BaseBroadCastIT {
List mqs = producer.getMessageQueue();
MessageQueueMsg mqMsgs = new MessageQueueMsg(mqs, msgSize);
producer.send(mqMsgs.getMsgsWithMQ());
-
- consumer1.getListener().waitForMessageConsume(producer.getAllMsgBody(), consumeTime);
- consumer2.getListener().waitForMessageConsume(producer.getAllMsgBody(), consumeTime);
+ consumer1.getListener().waitForMessageConsume(producer.getAllMsgBody(), broadcastConsumeTime);
+ consumer2.getListener().waitForMessageConsume(producer.getAllMsgBody(), broadcastConsumeTime);
assertThat(VerifyUtils.verifyOrder(((RMQOrderListener) consumer1.getListener()).getMsgs()))
.isEqualTo(true);
diff --git a/test/src/test/java/org/apache/rocketmq/test/client/consumer/filter/SqlFilterIT.java b/test/src/test/java/org/apache/rocketmq/test/client/consumer/filter/SqlFilterIT.java
index 115595d28d95d3349e50ab12eb144b253d32b328..15d91a1423ce14cb6fe6c48a7b406c09e1ceb3a3 100644
--- a/test/src/test/java/org/apache/rocketmq/test/client/consumer/filter/SqlFilterIT.java
+++ b/test/src/test/java/org/apache/rocketmq/test/client/consumer/filter/SqlFilterIT.java
@@ -67,7 +67,7 @@ public class SqlFilterIT extends BaseConf {
consumer.getListener().waitForMessageConsume(msgSize * 2, consumeTime);
assertThat(producer.getAllMsgBody())
.containsAllIn(VerifyUtils.getFilterdMessage(producer.getAllMsgBody(),
- consumer.getListener().getAllMsgBody()));
+ consumer.getListener().getAllMsgBody()));
assertThat(consumer.getListener().getAllMsgBody().size()).isEqualTo(msgSize * 2);
}
diff --git a/test/src/test/java/org/apache/rocketmq/test/client/producer/batch/BatchSendIT.java b/test/src/test/java/org/apache/rocketmq/test/client/producer/batch/BatchSendIT.java
index e372a1b2601b05a14b15ef9a6c903e009a632a31..6fb34af6d614b9f4decf9fdc3b954cc0f90673ca 100644
--- a/test/src/test/java/org/apache/rocketmq/test/client/producer/batch/BatchSendIT.java
+++ b/test/src/test/java/org/apache/rocketmq/test/client/producer/batch/BatchSendIT.java
@@ -77,7 +77,6 @@ public class BatchSendIT extends BaseConf {
}
}
-
@Test
public void testBatchSend_CheckProperties() throws Exception {
List messageList = new ArrayList<>();
@@ -91,7 +90,6 @@ public class BatchSendIT extends BaseConf {
message.setBody("body".getBytes());
messageList.add(message);
-
DefaultMQProducer producer = ProducerFactory.getRMQProducer(nsAddr);
SendResult sendResult = producer.send(messageList);
Assert.assertEquals(SendStatus.SEND_OK, sendResult.getSendStatus());
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java
index 409ea3322f213573edd79c731a2fdfa371ad5569..eb45de22f3e0427a560b4141b09acff77d79f6a9 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java
@@ -119,12 +119,14 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
}
@Override
- public MessageExt viewMessage(String offsetMsgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+ public MessageExt viewMessage(
+ String offsetMsgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
return defaultMQAdminExtImpl.viewMessage(offsetMsgId);
}
@Override
- public QueryResult queryMessage(String topic, String key, int maxNum, long begin, long end) throws MQClientException,
+ public QueryResult queryMessage(String topic, String key, int maxNum, long begin,
+ long end) throws MQClientException,
InterruptedException {
return defaultMQAdminExtImpl.queryMessage(topic, key, maxNum, begin, end);
}
@@ -140,7 +142,8 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
}
@Override
- public void updateBrokerConfig(String brokerAddr, Properties properties) throws RemotingConnectException, RemotingSendRequestException,
+ public void updateBrokerConfig(String brokerAddr,
+ Properties properties) throws RemotingConnectException, RemotingSendRequestException,
RemotingTimeoutException, UnsupportedEncodingException, InterruptedException, MQBrokerException {
defaultMQAdminExtImpl.updateBrokerConfig(brokerAddr, properties);
}
@@ -158,7 +161,8 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
}
@Override
- public void createAndUpdateSubscriptionGroupConfig(String addr, SubscriptionGroupConfig config) throws RemotingException,
+ public void createAndUpdateSubscriptionGroupConfig(String addr,
+ SubscriptionGroupConfig config) throws RemotingException,
MQBrokerException, InterruptedException, MQClientException {
defaultMQAdminExtImpl.createAndUpdateSubscriptionGroupConfig(addr, config);
}
@@ -174,7 +178,8 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
}
@Override
- public TopicStatsTable examineTopicStats(String topic) throws RemotingException, MQClientException, InterruptedException,
+ public TopicStatsTable examineTopicStats(
+ String topic) throws RemotingException, MQClientException, InterruptedException,
MQBrokerException {
return defaultMQAdminExtImpl.examineTopicStats(topic);
}
@@ -185,24 +190,28 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
}
@Override
- public TopicList fetchTopicsByCLuster(String clusterName) throws RemotingException, MQClientException, InterruptedException {
+ public TopicList fetchTopicsByCLuster(
+ String clusterName) throws RemotingException, MQClientException, InterruptedException {
return this.defaultMQAdminExtImpl.fetchTopicsByCLuster(clusterName);
}
@Override
- public KVTable fetchBrokerRuntimeStats(final String brokerAddr) throws RemotingConnectException, RemotingSendRequestException,
+ public KVTable fetchBrokerRuntimeStats(
+ final String brokerAddr) throws RemotingConnectException, RemotingSendRequestException,
RemotingTimeoutException, InterruptedException, MQBrokerException {
return this.defaultMQAdminExtImpl.fetchBrokerRuntimeStats(brokerAddr);
}
@Override
- public ConsumeStats examineConsumeStats(String consumerGroup) throws RemotingException, MQClientException, InterruptedException,
+ public ConsumeStats examineConsumeStats(
+ String consumerGroup) throws RemotingException, MQClientException, InterruptedException,
MQBrokerException {
return examineConsumeStats(consumerGroup, null);
}
@Override
- public ConsumeStats examineConsumeStats(String consumerGroup, String topic) throws RemotingException, MQClientException,
+ public ConsumeStats examineConsumeStats(String consumerGroup,
+ String topic) throws RemotingException, MQClientException,
InterruptedException, MQBrokerException {
return defaultMQAdminExtImpl.examineConsumeStats(consumerGroup, topic);
}
@@ -214,18 +223,21 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
}
@Override
- public TopicRouteData examineTopicRouteInfo(String topic) throws RemotingException, MQClientException, InterruptedException {
+ public TopicRouteData examineTopicRouteInfo(
+ String topic) throws RemotingException, MQClientException, InterruptedException {
return defaultMQAdminExtImpl.examineTopicRouteInfo(topic);
}
@Override
- public ConsumerConnection examineConsumerConnectionInfo(String consumerGroup) throws InterruptedException, MQBrokerException,
+ public ConsumerConnection examineConsumerConnectionInfo(
+ String consumerGroup) throws InterruptedException, MQBrokerException,
RemotingException, MQClientException {
return defaultMQAdminExtImpl.examineConsumerConnectionInfo(consumerGroup);
}
@Override
- public ProducerConnection examineProducerConnectionInfo(String producerGroup, final String topic) throws RemotingException,
+ public ProducerConnection examineProducerConnectionInfo(String producerGroup,
+ final String topic) throws RemotingException,
MQClientException, InterruptedException, MQBrokerException {
return defaultMQAdminExtImpl.examineProducerConnectionInfo(producerGroup, topic);
}
@@ -247,46 +259,54 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
}
@Override
- public String getKVConfig(String namespace, String key) throws RemotingException, MQClientException, InterruptedException {
+ public String getKVConfig(String namespace,
+ String key) throws RemotingException, MQClientException, InterruptedException {
return defaultMQAdminExtImpl.getKVConfig(namespace, key);
}
@Override
- public KVTable getKVListByNamespace(String namespace) throws RemotingException, MQClientException, InterruptedException {
+ public KVTable getKVListByNamespace(
+ String namespace) throws RemotingException, MQClientException, InterruptedException {
return defaultMQAdminExtImpl.getKVListByNamespace(namespace);
}
@Override
- public void deleteTopicInBroker(Set addrs, String topic) throws RemotingException, MQBrokerException, InterruptedException,
+ public void deleteTopicInBroker(Set addrs,
+ String topic) throws RemotingException, MQBrokerException, InterruptedException,
MQClientException {
defaultMQAdminExtImpl.deleteTopicInBroker(addrs, topic);
}
@Override
- public void deleteTopicInNameServer(Set addrs, String topic) throws RemotingException, MQBrokerException, InterruptedException,
+ public void deleteTopicInNameServer(Set addrs,
+ String topic) throws RemotingException, MQBrokerException, InterruptedException,
MQClientException {
defaultMQAdminExtImpl.deleteTopicInNameServer(addrs, topic);
}
@Override
- public void deleteSubscriptionGroup(String addr, String groupName) throws RemotingException, MQBrokerException, InterruptedException,
+ public void deleteSubscriptionGroup(String addr,
+ String groupName) throws RemotingException, MQBrokerException, InterruptedException,
MQClientException {
defaultMQAdminExtImpl.deleteSubscriptionGroup(addr, groupName);
}
@Override
- public void createAndUpdateKvConfig(String namespace, String key, String value) throws RemotingException, MQBrokerException,
+ public void createAndUpdateKvConfig(String namespace, String key,
+ String value) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException {
defaultMQAdminExtImpl.createAndUpdateKvConfig(namespace, key, value);
}
@Override
- public void deleteKvConfig(String namespace, String key) throws RemotingException, MQBrokerException, InterruptedException,
+ public void deleteKvConfig(String namespace,
+ String key) throws RemotingException, MQBrokerException, InterruptedException,
MQClientException {
defaultMQAdminExtImpl.deleteKvConfig(namespace, key);
}
- public List resetOffsetByTimestampOld(String consumerGroup, String topic, long timestamp, boolean force)
+ public List resetOffsetByTimestampOld(String consumerGroup, String topic, long timestamp,
+ boolean force)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
return defaultMQAdminExtImpl.resetOffsetByTimestampOld(consumerGroup, topic, timestamp, force);
}
@@ -297,49 +317,57 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
return resetOffsetByTimestamp(topic, group, timestamp, isForce, false);
}
- public Map resetOffsetByTimestamp(String topic, String group, long timestamp, boolean isForce, boolean isC)
+ public Map resetOffsetByTimestamp(String topic, String group, long timestamp, boolean isForce,
+ boolean isC)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
return defaultMQAdminExtImpl.resetOffsetByTimestamp(topic, group, timestamp, isForce, isC);
}
@Override
- public void resetOffsetNew(String consumerGroup, String topic, long timestamp) throws RemotingException, MQBrokerException,
+ public void resetOffsetNew(String consumerGroup, String topic,
+ long timestamp) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException {
this.defaultMQAdminExtImpl.resetOffsetNew(consumerGroup, topic, timestamp);
}
@Override
- public Map> getConsumeStatus(String topic, String group, String clientAddr) throws RemotingException,
+ public Map> getConsumeStatus(String topic, String group,
+ String clientAddr) throws RemotingException,
MQBrokerException, InterruptedException, MQClientException {
return defaultMQAdminExtImpl.getConsumeStatus(topic, group, clientAddr);
}
@Override
- public void createOrUpdateOrderConf(String key, String value, boolean isCluster) throws RemotingException, MQBrokerException,
+ public void createOrUpdateOrderConf(String key, String value,
+ boolean isCluster) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException {
defaultMQAdminExtImpl.createOrUpdateOrderConf(key, value, isCluster);
}
@Override
- public GroupList queryTopicConsumeByWho(String topic) throws InterruptedException, MQBrokerException, RemotingException,
+ public GroupList queryTopicConsumeByWho(
+ String topic) throws InterruptedException, MQBrokerException, RemotingException,
MQClientException {
return this.defaultMQAdminExtImpl.queryTopicConsumeByWho(topic);
}
@Override
- public List queryConsumeTimeSpan(final String topic, final String group) throws InterruptedException, MQBrokerException,
+ public List queryConsumeTimeSpan(final String topic,
+ final String group) throws InterruptedException, MQBrokerException,
RemotingException, MQClientException {
return this.defaultMQAdminExtImpl.queryConsumeTimeSpan(topic, group);
}
@Override
- public boolean cleanExpiredConsumerQueue(String cluster) throws RemotingConnectException, RemotingSendRequestException,
+ public boolean cleanExpiredConsumerQueue(
+ String cluster) throws RemotingConnectException, RemotingSendRequestException,
RemotingTimeoutException, MQClientException, InterruptedException {
return defaultMQAdminExtImpl.cleanExpiredConsumerQueue(cluster);
}
@Override
- public boolean cleanExpiredConsumerQueueByAddr(String addr) throws RemotingConnectException, RemotingSendRequestException,
+ public boolean cleanExpiredConsumerQueueByAddr(
+ String addr) throws RemotingConnectException, RemotingSendRequestException,
RemotingTimeoutException, MQClientException, InterruptedException {
return defaultMQAdminExtImpl.cleanExpiredConsumerQueueByAddr(addr);
}
@@ -357,7 +385,8 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
}
@Override
- public ConsumerRunningInfo getConsumerRunningInfo(String consumerGroup, String clientId, boolean jstack) throws RemotingException,
+ public ConsumerRunningInfo getConsumerRunningInfo(String consumerGroup, String clientId,
+ boolean jstack) throws RemotingException,
MQClientException, InterruptedException {
return defaultMQAdminExtImpl.getConsumerRunningInfo(consumerGroup, clientId, jstack);
}
@@ -369,25 +398,29 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
}
@Override
- public ConsumeMessageDirectlyResult consumeMessageDirectly(final String consumerGroup, final String clientId, final String topic,
+ public ConsumeMessageDirectlyResult consumeMessageDirectly(final String consumerGroup, final String clientId,
+ final String topic,
final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
return defaultMQAdminExtImpl.consumeMessageDirectly(consumerGroup, clientId, topic, msgId);
}
@Override
- public List messageTrackDetail(MessageExt msg) throws RemotingException, MQClientException, InterruptedException,
+ public List messageTrackDetail(
+ MessageExt msg) throws RemotingException, MQClientException, InterruptedException,
MQBrokerException {
return this.defaultMQAdminExtImpl.messageTrackDetail(msg);
}
@Override
- public void cloneGroupOffset(String srcGroup, String destGroup, String topic, boolean isOffline) throws RemotingException,
+ public void cloneGroupOffset(String srcGroup, String destGroup, String topic,
+ boolean isOffline) throws RemotingException,
MQClientException, InterruptedException, MQBrokerException {
this.defaultMQAdminExtImpl.cloneGroupOffset(srcGroup, destGroup, topic, isOffline);
}
@Override
- public BrokerStatsData viewBrokerStatsData(String brokerAddr, String statsName, String statsKey) throws RemotingConnectException,
+ public BrokerStatsData viewBrokerStatsData(String brokerAddr, String statsName,
+ String statsKey) throws RemotingConnectException,
RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException {
return this.defaultMQAdminExtImpl.viewBrokerStatsData(brokerAddr, statsName, statsKey);
}
@@ -406,7 +439,8 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
}
@Override
- public Set getTopicClusterList(final String topic) throws InterruptedException, MQBrokerException, MQClientException, RemotingException {
+ public Set getTopicClusterList(
+ final String topic) throws InterruptedException, MQBrokerException, MQClientException, RemotingException {
return this.defaultMQAdminExtImpl.getTopicClusterList(topic);
}
@@ -472,7 +506,8 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
}
@Override
- public QueryConsumeQueueResponseBody queryConsumeQueue(String brokerAddr, String topic, int queueId, long index, int count, String consumerGroup)
+ public QueryConsumeQueueResponseBody queryConsumeQueue(String brokerAddr, String topic, int queueId, long index,
+ int count, String consumerGroup)
throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException {
return this.defaultMQAdminExtImpl.queryConsumeQueue(
brokerAddr, topic, queueId, index, count, consumerGroup
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
index 157ae21eba3b49342215e2e79b5bb9ec319d9343..c93c40067a1f7d3798359a67a680decef5021d62 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
@@ -119,7 +119,7 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
if (!registerOK) {
this.serviceState = ServiceState.CREATE_JUST;
throw new MQClientException("The adminExt group[" + this.defaultMQAdminExt.getAdminExtGroup()
- + "] has created already, specifed another name please."//
+ + "] has created already, specifed another name please."
+ FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL), null);
}
@@ -132,8 +132,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
case RUNNING:
case START_FAILED:
case SHUTDOWN_ALREADY:
- throw new MQClientException("The AdminExt service state not OK, maybe started once, "//
- + this.serviceState//
+ throw new MQClientException("The AdminExt service state not OK, maybe started once, "
+ + this.serviceState
+ FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK), null);
default:
break;
@@ -160,7 +160,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public void updateBrokerConfig(String brokerAddr, Properties properties) throws RemotingConnectException, RemotingSendRequestException,
+ public void updateBrokerConfig(String brokerAddr,
+ Properties properties) throws RemotingConnectException, RemotingSendRequestException,
RemotingTimeoutException, UnsupportedEncodingException, InterruptedException, MQBrokerException {
this.mqClientInstance.getMQClientAPIImpl().updateBrokerConfig(brokerAddr, properties, timeoutMillis);
}
@@ -178,25 +179,25 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public void createAndUpdateSubscriptionGroupConfig(String addr, SubscriptionGroupConfig config) throws RemotingException,
+ public void createAndUpdateSubscriptionGroupConfig(String addr,
+ SubscriptionGroupConfig config) throws RemotingException,
MQBrokerException, InterruptedException, MQClientException {
this.mqClientInstance.getMQClientAPIImpl().createSubscriptionGroup(addr, config, timeoutMillis);
}
@Override
public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr, String group) {
- // TODO Auto-generated method stub
return null;
}
@Override
public TopicConfig examineTopicConfig(String addr, String topic) {
- // TODO Auto-generated method stub
return null;
}
@Override
- public TopicStatsTable examineTopicStats(String topic) throws RemotingException, MQClientException, InterruptedException,
+ public TopicStatsTable examineTopicStats(
+ String topic) throws RemotingException, MQClientException, InterruptedException,
MQBrokerException {
TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
TopicStatsTable topicStatsTable = new TopicStatsTable();
@@ -222,24 +223,28 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public TopicList fetchTopicsByCLuster(String clusterName) throws RemotingException, MQClientException, InterruptedException {
+ public TopicList fetchTopicsByCLuster(
+ String clusterName) throws RemotingException, MQClientException, InterruptedException {
return this.mqClientInstance.getMQClientAPIImpl().getTopicsByCluster(clusterName, timeoutMillis);
}
@Override
- public KVTable fetchBrokerRuntimeStats(final String brokerAddr) throws RemotingConnectException, RemotingSendRequestException,
+ public KVTable fetchBrokerRuntimeStats(
+ final String brokerAddr) throws RemotingConnectException, RemotingSendRequestException,
RemotingTimeoutException, InterruptedException, MQBrokerException {
return this.mqClientInstance.getMQClientAPIImpl().getBrokerRuntimeInfo(brokerAddr, timeoutMillis);
}
@Override
- public ConsumeStats examineConsumeStats(String consumerGroup) throws RemotingException, MQClientException, InterruptedException,
+ public ConsumeStats examineConsumeStats(
+ String consumerGroup) throws RemotingException, MQClientException, InterruptedException,
MQBrokerException {
return examineConsumeStats(consumerGroup, null);
}
@Override
- public ConsumeStats examineConsumeStats(String consumerGroup, String topic) throws RemotingException, MQClientException,
+ public ConsumeStats examineConsumeStats(String consumerGroup,
+ String topic) throws RemotingException, MQClientException,
InterruptedException, MQBrokerException {
String retryTopic = MixAll.getRetryTopic(consumerGroup);
TopicRouteData topicRouteData = this.examineTopicRouteInfo(retryTopic);
@@ -271,12 +276,14 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public TopicRouteData examineTopicRouteInfo(String topic) throws RemotingException, MQClientException, InterruptedException {
+ public TopicRouteData examineTopicRouteInfo(
+ String topic) throws RemotingException, MQClientException, InterruptedException {
return this.mqClientInstance.getMQClientAPIImpl().getTopicRouteInfoFromNameServer(topic, timeoutMillis);
}
@Override
- public MessageExt viewMessage(String topic, String msgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+ public MessageExt viewMessage(String topic,
+ String msgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
try {
MessageDecoder.decodeMessageId(msgId);
return this.viewMessage(msgId);
@@ -287,7 +294,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public ConsumerConnection examineConsumerConnectionInfo(String consumerGroup) throws InterruptedException, MQBrokerException,
+ public ConsumerConnection examineConsumerConnectionInfo(
+ String consumerGroup) throws InterruptedException, MQBrokerException,
RemotingException, MQClientException {
ConsumerConnection result = new ConsumerConnection();
String topic = MixAll.getRetryTopic(consumerGroup);
@@ -310,7 +318,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public ProducerConnection examineProducerConnectionInfo(String producerGroup, final String topic) throws RemotingException,
+ public ProducerConnection examineProducerConnectionInfo(String producerGroup,
+ final String topic) throws RemotingException,
MQClientException, InterruptedException, MQBrokerException {
ProducerConnection result = new ProducerConnection();
List brokers = this.examineTopicRouteInfo(topic).getBrokerDatas();
@@ -344,22 +353,23 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
@Override
public void putKVConfig(String namespace, String key, String value) {
- // TODO Auto-generated method stub
-
}
@Override
- public String getKVConfig(String namespace, String key) throws RemotingException, MQClientException, InterruptedException {
+ public String getKVConfig(String namespace,
+ String key) throws RemotingException, MQClientException, InterruptedException {
return this.mqClientInstance.getMQClientAPIImpl().getKVConfigValue(namespace, key, timeoutMillis);
}
@Override
- public KVTable getKVListByNamespace(String namespace) throws RemotingException, MQClientException, InterruptedException {
+ public KVTable getKVListByNamespace(
+ String namespace) throws RemotingException, MQClientException, InterruptedException {
return this.mqClientInstance.getMQClientAPIImpl().getKVListByNamespace(namespace, timeoutMillis);
}
@Override
- public void deleteTopicInBroker(Set addrs, String topic) throws RemotingException, MQBrokerException, InterruptedException,
+ public void deleteTopicInBroker(Set addrs,
+ String topic) throws RemotingException, MQBrokerException, InterruptedException,
MQClientException {
for (String addr : addrs) {
this.mqClientInstance.getMQClientAPIImpl().deleteTopicInBroker(addr, topic, timeoutMillis);
@@ -367,7 +377,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public void deleteTopicInNameServer(Set addrs, String topic) throws RemotingException, MQBrokerException, InterruptedException,
+ public void deleteTopicInNameServer(Set addrs,
+ String topic) throws RemotingException, MQBrokerException, InterruptedException,
MQClientException {
if (addrs == null) {
String ns = this.mqClientInstance.getMQClientAPIImpl().fetchNameServerAddr();
@@ -379,25 +390,29 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public void deleteSubscriptionGroup(String addr, String groupName) throws RemotingException, MQBrokerException, InterruptedException,
+ public void deleteSubscriptionGroup(String addr,
+ String groupName) throws RemotingException, MQBrokerException, InterruptedException,
MQClientException {
this.mqClientInstance.getMQClientAPIImpl().deleteSubscriptionGroup(addr, groupName, timeoutMillis);
}
@Override
- public void createAndUpdateKvConfig(String namespace, String key, String value) throws RemotingException, MQBrokerException,
+ public void createAndUpdateKvConfig(String namespace, String key,
+ String value) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException {
this.mqClientInstance.getMQClientAPIImpl().putKVConfigValue(namespace, key, value, timeoutMillis);
}
@Override
- public void deleteKvConfig(String namespace, String key) throws RemotingException, MQBrokerException, InterruptedException,
+ public void deleteKvConfig(String namespace,
+ String key) throws RemotingException, MQBrokerException, InterruptedException,
MQClientException {
this.mqClientInstance.getMQClientAPIImpl().deleteKVConfigValue(namespace, key, timeoutMillis);
}
@Override
- public List resetOffsetByTimestampOld(String consumerGroup, String topic, long timestamp, boolean force)
+ public List resetOffsetByTimestampOld(String consumerGroup, String topic, long timestamp,
+ boolean force)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
List rollbackStatsList = new ArrayList();
@@ -448,7 +463,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public void resetOffsetNew(String consumerGroup, String topic, long timestamp) throws RemotingException, MQBrokerException,
+ public void resetOffsetNew(String consumerGroup, String topic,
+ long timestamp) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException {
try {
this.resetOffsetByTimestamp(topic, consumerGroup, timestamp, true);
@@ -461,7 +477,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
}
- public Map resetOffsetByTimestamp(String topic, String group, long timestamp, boolean isForce, boolean isC)
+ public Map resetOffsetByTimestamp(String topic, String group, long timestamp, boolean isForce,
+ boolean isC)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
List brokerDatas = topicRouteData.getBrokerDatas();
@@ -482,7 +499,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
return allOffsetTable;
}
- private RollbackStats resetOffsetConsumeOffset(String brokerAddr, String consumeGroup, MessageQueue queue, OffsetWrapper offsetWrapper,
+ private RollbackStats resetOffsetConsumeOffset(String brokerAddr, String consumeGroup, MessageQueue queue,
+ OffsetWrapper offsetWrapper,
long timestamp, boolean force) throws RemotingException, InterruptedException, MQBrokerException {
long resetOffset;
if (timestamp == -1) {
@@ -515,7 +533,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public Map> getConsumeStatus(String topic, String group, String clientAddr) throws RemotingException,
+ public Map> getConsumeStatus(String topic, String group,
+ String clientAddr) throws RemotingException,
MQBrokerException, InterruptedException, MQClientException {
TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
List brokerDatas = topicRouteData.getBrokerDatas();
@@ -529,7 +548,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
return Collections.EMPTY_MAP;
}
- public void createOrUpdateOrderConf(String key, String value, boolean isCluster) throws RemotingException, MQBrokerException,
+ public void createOrUpdateOrderConf(String key, String value,
+ boolean isCluster) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException {
if (isCluster) {
@@ -568,7 +588,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public GroupList queryTopicConsumeByWho(String topic) throws InterruptedException, MQBrokerException, RemotingException,
+ public GroupList queryTopicConsumeByWho(
+ String topic) throws InterruptedException, MQBrokerException, RemotingException,
MQClientException {
TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
@@ -585,7 +606,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public List queryConsumeTimeSpan(final String topic, final String group) throws InterruptedException, MQBrokerException,
+ public List queryConsumeTimeSpan(final String topic,
+ final String group) throws InterruptedException, MQBrokerException,
RemotingException, MQClientException {
List spanSet = new ArrayList();
TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
@@ -599,7 +621,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public boolean cleanExpiredConsumerQueue(String cluster) throws RemotingConnectException, RemotingSendRequestException,
+ public boolean cleanExpiredConsumerQueue(
+ String cluster) throws RemotingConnectException, RemotingSendRequestException,
RemotingTimeoutException, MQClientException, InterruptedException {
boolean result = false;
try {
@@ -618,7 +641,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
return result;
}
- public boolean cleanExpiredConsumerQueueByCluster(ClusterInfo clusterInfo, String cluster) throws RemotingConnectException,
+ public boolean cleanExpiredConsumerQueueByCluster(ClusterInfo clusterInfo,
+ String cluster) throws RemotingConnectException,
RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException {
boolean result = false;
String[] addrs = clusterInfo.retrieveAllAddrByCluster(cluster);
@@ -629,7 +653,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public boolean cleanExpiredConsumerQueueByAddr(String addr) throws RemotingConnectException, RemotingSendRequestException,
+ public boolean cleanExpiredConsumerQueueByAddr(
+ String addr) throws RemotingConnectException, RemotingSendRequestException,
RemotingTimeoutException, MQClientException, InterruptedException {
boolean result = mqClientInstance.getMQClientAPIImpl().cleanExpiredConsumeQueue(addr, timeoutMillis);
log.warn("clean expired ConsumeQueue on target " + addr + " broker " + result);
@@ -675,7 +700,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public ConsumerRunningInfo getConsumerRunningInfo(String consumerGroup, String clientId, boolean jstack) throws RemotingException,
+ public ConsumerRunningInfo getConsumerRunningInfo(String consumerGroup, String clientId,
+ boolean jstack) throws RemotingException,
MQClientException, InterruptedException {
String topic = MixAll.RETRY_GROUP_TOPIC_PREFIX + consumerGroup;
TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
@@ -702,7 +728,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public ConsumeMessageDirectlyResult consumeMessageDirectly(final String consumerGroup, final String clientId, final String topic,
+ public ConsumeMessageDirectlyResult consumeMessageDirectly(final String consumerGroup, final String clientId,
+ final String topic,
final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
MessageExt msg = this.viewMessage(topic, msgId);
if (msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX) == null) {
@@ -716,7 +743,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public List messageTrackDetail(MessageExt msg) throws RemotingException, MQClientException, InterruptedException,
+ public List messageTrackDetail(
+ MessageExt msg) throws RemotingException, MQClientException, InterruptedException,
MQBrokerException {
List result = new ArrayList();
@@ -798,7 +826,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
return result;
}
- public boolean consumed(final MessageExt msg, final String group) throws RemotingException, MQClientException, InterruptedException,
+ public boolean consumed(final MessageExt msg,
+ final String group) throws RemotingException, MQClientException, InterruptedException,
MQBrokerException {
ConsumeStats cstats = this.examineConsumeStats(group);
@@ -826,7 +855,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public void cloneGroupOffset(String srcGroup, String destGroup, String topic, boolean isOffline) throws RemotingException,
+ public void cloneGroupOffset(String srcGroup, String destGroup, String topic,
+ boolean isOffline) throws RemotingException,
MQClientException, InterruptedException, MQBrokerException {
String retryTopic = MixAll.getRetryTopic(srcGroup);
TopicRouteData topicRouteData = this.examineTopicRouteInfo(retryTopic);
@@ -840,7 +870,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public BrokerStatsData viewBrokerStatsData(String brokerAddr, String statsName, String statsKey) throws RemotingConnectException,
+ public BrokerStatsData viewBrokerStatsData(String brokerAddr, String statsName,
+ String statsKey) throws RemotingConnectException,
RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException {
return this.mqClientInstance.getMQClientAPIImpl().viewBrokerStatsData(brokerAddr, statsName, statsKey, timeoutMillis);
}
@@ -859,7 +890,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public Set getTopicClusterList(final String topic) throws InterruptedException, MQBrokerException, MQClientException,
+ public Set getTopicClusterList(
+ final String topic) throws InterruptedException, MQBrokerException, MQClientException,
RemotingException {
Set clusterSet = new HashSet();
ClusterInfo clusterInfo = examineBrokerClusterInfo();
@@ -877,13 +909,15 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public SubscriptionGroupWrapper getAllSubscriptionGroup(final String brokerAddr, long timeoutMillis) throws InterruptedException,
+ public SubscriptionGroupWrapper getAllSubscriptionGroup(final String brokerAddr,
+ long timeoutMillis) throws InterruptedException,
RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQBrokerException {
return this.mqClientInstance.getMQClientAPIImpl().getAllSubscriptionGroup(brokerAddr, timeoutMillis);
}
@Override
- public TopicConfigSerializeWrapper getAllTopicGroup(final String brokerAddr, long timeoutMillis) throws InterruptedException,
+ public TopicConfigSerializeWrapper getAllTopicGroup(final String brokerAddr,
+ long timeoutMillis) throws InterruptedException,
RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQBrokerException {
return this.mqClientInstance.getMQClientAPIImpl().getAllTopicConfig(brokerAddr, timeoutMillis);
}
@@ -919,12 +953,14 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public MessageExt viewMessage(String msgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+ public MessageExt viewMessage(
+ String msgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
return this.mqClientInstance.getMQAdminImpl().viewMessage(msgId);
}
@Override
- public QueryResult queryMessage(String topic, String key, int maxNum, long begin, long end) throws MQClientException,
+ public QueryResult queryMessage(String topic, String key, int maxNum, long begin,
+ long end) throws MQClientException,
InterruptedException {
return this.mqClientInstance.getMQAdminImpl().queryMessage(topic, key, maxNum, begin, end);
}
@@ -957,7 +993,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
- public QueryConsumeQueueResponseBody queryConsumeQueue(String brokerAddr, String topic, int queueId, long index, int count, String consumerGroup)
+ public QueryConsumeQueueResponseBody queryConsumeQueue(String brokerAddr, String topic, int queueId, long index,
+ int count, String consumerGroup)
throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException {
return this.mqClientInstance.getMQClientAPIImpl().queryConsumeQueue(
brokerAddr, topic, queueId, index, count, consumerGroup, timeoutMillis
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java b/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java
index 82add92b2d60d721fdd2a4029e4d35a1f16a26be..16b4427575faaa021ba97ea1828ebdab95f7bc8c 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java
@@ -64,42 +64,51 @@ public interface MQAdminExt extends MQAdmin {
Properties getBrokerConfig(final String brokerAddr) throws RemotingConnectException,
RemotingSendRequestException, RemotingTimeoutException, UnsupportedEncodingException, InterruptedException, MQBrokerException;
- void createAndUpdateTopicConfig(final String addr, final TopicConfig config) throws RemotingException, MQBrokerException,
+ void createAndUpdateTopicConfig(final String addr,
+ final TopicConfig config) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException;
- void createAndUpdateSubscriptionGroupConfig(final String addr, final SubscriptionGroupConfig config) throws RemotingException,
+ void createAndUpdateSubscriptionGroupConfig(final String addr,
+ final SubscriptionGroupConfig config) throws RemotingException,
MQBrokerException, InterruptedException, MQClientException;
SubscriptionGroupConfig examineSubscriptionGroupConfig(final String addr, final String group);
TopicConfig examineTopicConfig(final String addr, final String topic);
- TopicStatsTable examineTopicStats(final String topic) throws RemotingException, MQClientException, InterruptedException,
+ TopicStatsTable examineTopicStats(
+ final String topic) throws RemotingException, MQClientException, InterruptedException,
MQBrokerException;
TopicList fetchAllTopicList() throws RemotingException, MQClientException, InterruptedException;
- TopicList fetchTopicsByCLuster(String clusterName) throws RemotingException, MQClientException, InterruptedException;
+ TopicList fetchTopicsByCLuster(
+ String clusterName) throws RemotingException, MQClientException, InterruptedException;
- KVTable fetchBrokerRuntimeStats(final String brokerAddr) throws RemotingConnectException, RemotingSendRequestException,
+ KVTable fetchBrokerRuntimeStats(
+ final String brokerAddr) throws RemotingConnectException, RemotingSendRequestException,
RemotingTimeoutException, InterruptedException, MQBrokerException;
- ConsumeStats examineConsumeStats(final String consumerGroup) throws RemotingException, MQClientException, InterruptedException,
+ ConsumeStats examineConsumeStats(
+ final String consumerGroup) throws RemotingException, MQClientException, InterruptedException,
MQBrokerException;
- ConsumeStats examineConsumeStats(final String consumerGroup, final String topic) throws RemotingException, MQClientException,
+ ConsumeStats examineConsumeStats(final String consumerGroup,
+ final String topic) throws RemotingException, MQClientException,
InterruptedException, MQBrokerException;
ClusterInfo examineBrokerClusterInfo() throws InterruptedException, MQBrokerException, RemotingTimeoutException,
RemotingSendRequestException, RemotingConnectException;
- TopicRouteData examineTopicRouteInfo(final String topic) throws RemotingException, MQClientException, InterruptedException;
+ TopicRouteData examineTopicRouteInfo(
+ final String topic) throws RemotingException, MQClientException, InterruptedException;
ConsumerConnection examineConsumerConnectionInfo(final String consumerGroup) throws RemotingConnectException,
RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException, RemotingException,
MQClientException;
- ProducerConnection examineProducerConnectionInfo(final String producerGroup, final String topic) throws RemotingException,
+ ProducerConnection examineProducerConnectionInfo(final String producerGroup,
+ final String topic) throws RemotingException,
MQClientException, InterruptedException, MQBrokerException;
List getNameServerAddressList();
@@ -109,20 +118,24 @@ public interface MQAdminExt extends MQAdmin {
void putKVConfig(final String namespace, final String key, final String value);
- String getKVConfig(final String namespace, final String key) throws RemotingException, MQClientException, InterruptedException;
+ String getKVConfig(final String namespace,
+ final String key) throws RemotingException, MQClientException, InterruptedException;
- KVTable getKVListByNamespace(final String namespace) throws RemotingException, MQClientException, InterruptedException;
+ KVTable getKVListByNamespace(
+ final String namespace) throws RemotingException, MQClientException, InterruptedException;
void deleteTopicInBroker(final Set addrs, final String topic) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException;
- void deleteTopicInNameServer(final Set addrs, final String topic) throws RemotingException, MQBrokerException,
+ void deleteTopicInNameServer(final Set addrs,
+ final String topic) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException;
void deleteSubscriptionGroup(final String addr, String groupName) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException;
- void createAndUpdateKvConfig(String namespace, String key, String value) throws RemotingException, MQBrokerException,
+ void createAndUpdateKvConfig(String namespace, String key,
+ String value) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException;
void deleteKvConfig(String namespace, String key) throws RemotingException, MQBrokerException, InterruptedException,
@@ -137,16 +150,19 @@ public interface MQAdminExt extends MQAdmin {
void resetOffsetNew(String consumerGroup, String topic, long timestamp) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException;
- Map> getConsumeStatus(String topic, String group, String clientAddr) throws RemotingException,
+ Map> getConsumeStatus(String topic, String group,
+ String clientAddr) throws RemotingException,
MQBrokerException, InterruptedException, MQClientException;
- void createOrUpdateOrderConf(String key, String value, boolean isCluster) throws RemotingException, MQBrokerException,
+ void createOrUpdateOrderConf(String key, String value,
+ boolean isCluster) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException;
GroupList queryTopicConsumeByWho(final String topic) throws RemotingConnectException, RemotingSendRequestException,
RemotingTimeoutException, InterruptedException, MQBrokerException, RemotingException, MQClientException;
- List queryConsumeTimeSpan(final String topic, final String group) throws InterruptedException, MQBrokerException,
+ List queryConsumeTimeSpan(final String topic,
+ final String group) throws InterruptedException, MQBrokerException,
RemotingException, MQClientException;
boolean cleanExpiredConsumerQueue(String cluster) throws RemotingConnectException, RemotingSendRequestException,
@@ -173,7 +189,8 @@ public interface MQAdminExt extends MQAdmin {
String topic,
String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException;
- List messageTrackDetail(MessageExt msg) throws RemotingException, MQClientException, InterruptedException,
+ List messageTrackDetail(
+ MessageExt msg) throws RemotingException, MQClientException, InterruptedException,
MQBrokerException;
void cloneGroupOffset(String srcGroup, String destGroup, String topic, boolean isOffline) throws RemotingException,
@@ -190,7 +207,8 @@ public interface MQAdminExt extends MQAdmin {
long timeoutMillis) throws RemotingConnectException, RemotingSendRequestException,
RemotingTimeoutException, MQClientException, InterruptedException;
- Set getTopicClusterList(final String topic) throws InterruptedException, MQBrokerException, MQClientException, RemotingException;
+ Set getTopicClusterList(
+ final String topic) throws InterruptedException, MQBrokerException, MQClientException, RemotingException;
SubscriptionGroupWrapper getAllSubscriptionGroup(final String brokerAddr,
long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException,
@@ -209,18 +227,9 @@ public interface MQAdminExt extends MQAdmin {
* Command Code : RequestCode.UPDATE_NAMESRV_CONFIG
*
*
If param(nameServers) is null or empty, will use name servers from ns!
- *
- * @param properties
- * @param nameServers
- * @throws InterruptedException
- * @throws RemotingConnectException
- * @throws UnsupportedEncodingException
- * @throws RemotingSendRequestException
- * @throws RemotingTimeoutException
- * @throws MQClientException
- * @throws MQBrokerException
*/
- void updateNameServerConfig(final Properties properties, final List nameServers) throws InterruptedException, RemotingConnectException,
+ void updateNameServerConfig(final Properties properties,
+ final List nameServers) throws InterruptedException, RemotingConnectException,
UnsupportedEncodingException, RemotingSendRequestException, RemotingTimeoutException,
MQClientException, MQBrokerException;
@@ -230,14 +239,7 @@ public interface MQAdminExt extends MQAdmin {
* Command Code : RequestCode.GET_NAMESRV_CONFIG
*
If param(nameServers) is null or empty, will use name servers from ns!
*
- * @param nameServers
* @return The fetched name server config
- * @throws InterruptedException
- * @throws RemotingTimeoutException
- * @throws RemotingSendRequestException
- * @throws RemotingConnectException
- * @throws MQClientException
- * @throws UnsupportedEncodingException
*/
Map getNameServerConfig(final List nameServers) throws InterruptedException,
RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException,
@@ -252,15 +254,9 @@ public interface MQAdminExt extends MQAdmin {
* @param index start offset
* @param count how many
* @param consumerGroup group
- * @return
- * @throws InterruptedException
- * @throws RemotingTimeoutException
- * @throws RemotingSendRequestException
- * @throws RemotingConnectException
- * @throws MQClientException
*/
QueryConsumeQueueResponseBody queryConsumeQueue(final String brokerAddr,
- final String topic, final int queueId,
- final long index, final int count, final String consumerGroup)
+ final String topic, final int queueId,
+ final long index, final int count, final String consumerGroup)
throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException, MQClientException;
}
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/broker/GetBrokerConfigCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/broker/GetBrokerConfigCommand.java
index 919f673e29b0ff1e2a07cddd4aa63d1252b5ce1c..11a36048328f4e40ebcfc0decb3092cd19f4238c 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/broker/GetBrokerConfigCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/broker/GetBrokerConfigCommand.java
@@ -60,7 +60,8 @@ public class GetBrokerConfigCommand implements SubCommand {
}
@Override
- public void execute(final CommandLine commandLine, final Options options, final RPCHook rpcHook) throws SubCommandException {
+ public void execute(final CommandLine commandLine, final Options options,
+ final RPCHook rpcHook) throws SubCommandException {
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/cluster/ClusterListSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/cluster/ClusterListSubCommand.java
index c7b8ac546757af23e62addf338332e72c91ed864..6a0cd71c177b5f2ebffef15d6c0402bbaedca59d 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/cluster/ClusterListSubCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/cluster/ClusterListSubCommand.java
@@ -61,7 +61,8 @@ public class ClusterListSubCommand implements SubCommand {
}
@Override
- public void execute(final CommandLine commandLine, final Options options, RPCHook rpcHook) throws SubCommandException {
+ public void execute(final CommandLine commandLine, final Options options,
+ RPCHook rpcHook) throws SubCommandException {
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
@@ -164,7 +165,8 @@ public class ClusterListSubCommand implements SubCommand {
}
}
- private void printClusterBaseInfo(final DefaultMQAdminExt defaultMQAdminExt) throws RemotingConnectException, RemotingTimeoutException,
+ private void printClusterBaseInfo(
+ final DefaultMQAdminExt defaultMQAdminExt) throws RemotingConnectException, RemotingTimeoutException,
RemotingSendRequestException, InterruptedException, MQBrokerException {
ClusterInfo clusterInfoSerializeWrapper = defaultMQAdminExt.examineBrokerClusterInfo();
@@ -253,16 +255,16 @@ public class ClusterListSubCommand implements SubCommand {
}
System.out.printf("%-16s %-22s %-4s %-22s %-16s %19s %19s %10s %5s %6s%n",
- clusterName,
- brokerName,
- next1.getKey(),
- next1.getValue(),
- version,
- String.format("%9.2f(%s,%sms)", in, sendThreadPoolQueueSize, sendThreadPoolQueueHeadWaitTimeMills),
- String.format("%9.2f(%s,%sms)", out, pullThreadPoolQueueSize, pullThreadPoolQueueHeadWaitTimeMills),
- pageCacheLockTimeMills,
- String.format("%2.2f", hour),
- String.format("%.4f", space)
+ clusterName,
+ brokerName,
+ next1.getKey(),
+ next1.getValue(),
+ version,
+ String.format("%9.2f(%s,%sms)", in, sendThreadPoolQueueSize, sendThreadPoolQueueHeadWaitTimeMills),
+ String.format("%9.2f(%s,%sms)", out, pullThreadPoolQueueSize, pullThreadPoolQueueHeadWaitTimeMills),
+ pageCacheLockTimeMills,
+ String.format("%2.2f", hour),
+ String.format("%.4f", space)
);
}
}
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/consumer/UpdateSubGroupSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/consumer/UpdateSubGroupSubCommand.java
index 910eb1c032c7d7bad16a413c96f1f9bb69f7cb94..7316526be01f4855e11ea9efd749f79628bedb17 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/consumer/UpdateSubGroupSubCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/consumer/UpdateSubGroupSubCommand.java
@@ -90,7 +90,8 @@ public class UpdateSubGroupSubCommand implements SubCommand {
}
@Override
- public void execute(final CommandLine commandLine, final Options options, RPCHook rpcHook) throws SubCommandException {
+ public void execute(final CommandLine commandLine, final Options options,
+ RPCHook rpcHook) throws SubCommandException {
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommond.java b/tools/src/main/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommond.java
index 64f634e92d07f3d36bdbc1f3f19204be22c80524..d233b65aa948863dabe9e5054079819a1b1ca162 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommond.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/message/DecodeMessageIdCommond.java
@@ -45,7 +45,8 @@ public class DecodeMessageIdCommond implements SubCommand {
}
@Override
- public void execute(final CommandLine commandLine, final Options options, RPCHook rpcHook) throws SubCommandException {
+ public void execute(final CommandLine commandLine, final Options options,
+ RPCHook rpcHook) throws SubCommandException {
String messageId = commandLine.getOptionValue('i').trim();
try {
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/message/PrintMessageByQueueCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/message/PrintMessageByQueueCommand.java
index ac51267161c399d5c03129b88226a586640560c2..46c5f74735265066c96050579db1ec9efbfb10a2 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/message/PrintMessageByQueueCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/message/PrintMessageByQueueCommand.java
@@ -52,7 +52,8 @@ public class PrintMessageByQueueCommand implements SubCommand {
return timestamp;
}
- private static void calculateByTag(final List msgs, final Map tagCalmap, final boolean calByTag) {
+ private static void calculateByTag(final List msgs, final Map tagCalmap,
+ final boolean calByTag) {
if (!calByTag)
return;
@@ -85,7 +86,8 @@ public class PrintMessageByQueueCommand implements SubCommand {
}
}
- public static void printMessage(final List msgs, final String charsetName, boolean printMsg, boolean printBody) {
+ public static void printMessage(final List msgs, final String charsetName, boolean printMsg,
+ boolean printBody) {
if (!printMsg)
return;
@@ -162,11 +164,11 @@ public class PrintMessageByQueueCommand implements SubCommand {
String charsetName =
!commandLine.hasOption('c') ? "UTF-8" : commandLine.getOptionValue('c').trim();
boolean printMsg =
- commandLine.hasOption('p') && Boolean.parseBoolean(commandLine.getOptionValue('p').trim());
+ commandLine.hasOption('p') && Boolean.parseBoolean(commandLine.getOptionValue('p').trim());
boolean printBody =
- commandLine.hasOption('d') && Boolean.parseBoolean(commandLine.getOptionValue('d').trim());
+ commandLine.hasOption('d') && Boolean.parseBoolean(commandLine.getOptionValue('d').trim());
boolean calByTag =
- commandLine.hasOption('f') && Boolean.parseBoolean(commandLine.getOptionValue('f').trim());
+ commandLine.hasOption('f') && Boolean.parseBoolean(commandLine.getOptionValue('f').trim());
String subExpression =
!commandLine.hasOption('s') ? "*" : commandLine.getOptionValue('s').trim();
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/message/PrintMessageSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/message/PrintMessageSubCommand.java
index ac485199c04c39b5e45bbc22e86248cb3f2c59fd..d46d6fb629d610d150a2b1ea9685f9f2897adaa2 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/message/PrintMessageSubCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/message/PrintMessageSubCommand.java
@@ -51,7 +51,6 @@ public class PrintMessageSubCommand implements SubCommand {
System.out.printf("MSGID: %s %s BODY: %s%n", msg.getMsgId(), msg.toString(),
printBody ? new String(msg.getBody(), charsetName) : "NOT PRINT BODY");
} catch (UnsupportedEncodingException e) {
- //
}
}
}
@@ -108,10 +107,10 @@ public class PrintMessageSubCommand implements SubCommand {
try {
String topic = commandLine.getOptionValue('t').trim();
- String charsetName = //
+ String charsetName =
!commandLine.hasOption('c') ? "UTF-8" : commandLine.getOptionValue('c').trim();
- String subExpression = //
+ String subExpression =
!commandLine.hasOption('s') ? "*" : commandLine.getOptionValue('s').trim();
boolean printBody = !commandLine.hasOption('d') || Boolean.parseBoolean(commandLine.getOptionValue('d').trim());
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/message/QueryMsgByIdSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/message/QueryMsgByIdSubCommand.java
index 05ae003084fea4968a09ceec70636234d052f0f3..39abbc9c4c0bf754412a74e19485b8706e9d6f38 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/message/QueryMsgByIdSubCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/message/QueryMsgByIdSubCommand.java
@@ -262,7 +262,8 @@ public class QueryMsgByIdSubCommand implements SubCommand {
}
}
- private void pushMsg(final DefaultMQAdminExt defaultMQAdminExt, final String consumerGroup, final String clientId, final String msgId) {
+ private void pushMsg(final DefaultMQAdminExt defaultMQAdminExt, final String consumerGroup, final String clientId,
+ final String msgId) {
try {
ConsumeMessageDirectlyResult result =
defaultMQAdminExt.consumeMessageDirectly(consumerGroup, clientId, msgId);
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/message/QueryMsgByUniqueKeySubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/message/QueryMsgByUniqueKeySubCommand.java
index 5c93ad7985b12f247f4d9aeb7bea2dc2c51dfc20..0103b50b35db83ff6c4c76bf4bd169658a9fa522 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/message/QueryMsgByUniqueKeySubCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/message/QueryMsgByUniqueKeySubCommand.java
@@ -39,7 +39,8 @@ import org.apache.rocketmq.tools.command.SubCommandException;
public class QueryMsgByUniqueKeySubCommand implements SubCommand {
- public static void queryById(final DefaultMQAdminExt admin, final String topic, final String msgId) throws MQClientException,
+ public static void queryById(final DefaultMQAdminExt admin, final String topic,
+ final String msgId) throws MQClientException,
RemotingException, MQBrokerException, InterruptedException, IOException {
MessageExt msg = admin.viewMessage(topic, msgId);
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/namesrv/GetNamesrvConfigCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/namesrv/GetNamesrvConfigCommand.java
index ce63616dce10ad56b8cb3e3b67b79ddc7f3b2b0c..22ce867b227e70f657348905b1989881785a285f 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/namesrv/GetNamesrvConfigCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/namesrv/GetNamesrvConfigCommand.java
@@ -46,7 +46,8 @@ public class GetNamesrvConfigCommand implements SubCommand {
}
@Override
- public void execute(final CommandLine commandLine, final Options options, final RPCHook rpcHook) throws SubCommandException {
+ public void execute(final CommandLine commandLine, final Options options,
+ final RPCHook rpcHook) throws SubCommandException {
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
try {
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/namesrv/UpdateNamesrvConfigCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/namesrv/UpdateNamesrvConfigCommand.java
index 250937027bb4e2ee089b680a539c6ed040b24531..1bb5b380292309fbcf7d4836ddd8db464aa5408e 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/namesrv/UpdateNamesrvConfigCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/namesrv/UpdateNamesrvConfigCommand.java
@@ -53,7 +53,8 @@ public class UpdateNamesrvConfigCommand implements SubCommand {
}
@Override
- public void execute(final CommandLine commandLine, final Options options, final RPCHook rpcHook) throws SubCommandException {
+ public void execute(final CommandLine commandLine, final Options options,
+ final RPCHook rpcHook) throws SubCommandException {
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
try {
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/queue/QueryConsumeQueueCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/queue/QueryConsumeQueueCommand.java
index 611addd54566b344efe4f63d7372d220fe243330..24767cc46f7fe86b5a1820bd039b745acf4a8794 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/queue/QueryConsumeQueueCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/queue/QueryConsumeQueueCommand.java
@@ -36,7 +36,7 @@ public class QueryConsumeQueueCommand implements SubCommand {
QueryConsumeQueueCommand cmd = new QueryConsumeQueueCommand();
Options options = ServerUtil.buildCommandlineOptions(new Options());
- String[] subargs = new String[]{"-t TopicTest", "-q 0", "-i 6447", "-b 100.81.165.119:10911"};
+ String[] subargs = new String[] {"-t TopicTest", "-q 0", "-i 6447", "-b 100.81.165.119:10911"};
final CommandLine commandLine =
ServerUtil.parseCmdLine("mqadmin " + cmd.commandName(), subargs, cmd.buildCommandlineOptions(options),
new PosixParser());
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicClusterSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicClusterSubCommand.java
index 7e2c7e3ae94cd0c9d5426bd383e7802d55fe21c5..a658b9829897a0b94154d97a07ee53d6f65e37f8 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicClusterSubCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicClusterSubCommand.java
@@ -46,7 +46,8 @@ public class TopicClusterSubCommand implements SubCommand {
}
@Override
- public void execute(final CommandLine commandLine, final Options options, RPCHook rpcHook) throws SubCommandException {
+ public void execute(final CommandLine commandLine, final Options options,
+ RPCHook rpcHook) throws SubCommandException {
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
String topic = commandLine.getOptionValue('t').trim();
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicListSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicListSubCommand.java
index 70f296ba9b4cd56db39331616bf60e8f63e91892..56aea2e8084460d694740019752dba55d9bdff7d 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicListSubCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicListSubCommand.java
@@ -57,7 +57,8 @@ public class TopicListSubCommand implements SubCommand {
}
@Override
- public void execute(final CommandLine commandLine, final Options options, RPCHook rpcHook) throws SubCommandException {
+ public void execute(final CommandLine commandLine, final Options options,
+ RPCHook rpcHook) throws SubCommandException {
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicRouteSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicRouteSubCommand.java
index d4efd900349cf9d1871a6c2d27d74982ac2e7943..c770db01d427c9e33c17a8a6cc20a76ed6760dfc 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicRouteSubCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicRouteSubCommand.java
@@ -47,7 +47,8 @@ public class TopicRouteSubCommand implements SubCommand {
}
@Override
- public void execute(final CommandLine commandLine, final Options options, RPCHook rpcHook) throws SubCommandException {
+ public void execute(final CommandLine commandLine, final Options options,
+ RPCHook rpcHook) throws SubCommandException {
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicStatusSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicStatusSubCommand.java
index f4ca80e530937891e74ca48c8f431db3883cb846..55440821b0e60b757d65262ba7dcace668ca13c3 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicStatusSubCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicStatusSubCommand.java
@@ -52,7 +52,8 @@ public class TopicStatusSubCommand implements SubCommand {
}
@Override
- public void execute(final CommandLine commandLine, final Options options, RPCHook rpcHook) throws SubCommandException {
+ public void execute(final CommandLine commandLine, final Options options,
+ RPCHook rpcHook) throws SubCommandException {
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateOrderConfCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateOrderConfCommand.java
index de26255d2ada57b0aff05d70b53decda660b9c0b..bebc646b43ecce92bd8a77ca7a74374ea8ce8c3d 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateOrderConfCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateOrderConfCommand.java
@@ -57,7 +57,8 @@ public class UpdateOrderConfCommand implements SubCommand {
}
@Override
- public void execute(final CommandLine commandLine, final Options options, RPCHook rpcHook) throws SubCommandException {
+ public void execute(final CommandLine commandLine, final Options options,
+ RPCHook rpcHook) throws SubCommandException {
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateTopicPermSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateTopicPermSubCommand.java
index 764c72f5bf43d76b01801f6ed96d16263614dc81..69420913e46c9b324c2fcb6161570974cf234e1f 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateTopicPermSubCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateTopicPermSubCommand.java
@@ -65,7 +65,8 @@ public class UpdateTopicPermSubCommand implements SubCommand {
}
@Override
- public void execute(final CommandLine commandLine, final Options options, RPCHook rpcHook) throws SubCommandException {
+ public void execute(final CommandLine commandLine, final Options options,
+ RPCHook rpcHook) throws SubCommandException {
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
try {
diff --git a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateTopicSubCommand.java b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateTopicSubCommand.java
index 4c84cd47e0c0a8fb8332a4423ee51288510320dc..9fdfcb38b244837554aec93579aef2dcb4abb737 100644
--- a/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateTopicSubCommand.java
+++ b/tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateTopicSubCommand.java
@@ -83,7 +83,8 @@ public class UpdateTopicSubCommand implements SubCommand {
}
@Override
- public void execute(final CommandLine commandLine, final Options options, RPCHook rpcHook) throws SubCommandException {
+ public void execute(final CommandLine commandLine, final Options options,
+ RPCHook rpcHook) throws SubCommandException {
DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
diff --git a/tools/src/test/java/org/apache/rocketmq/tools/command/namesrv/GetNamesrvConfigCommandTest.java b/tools/src/test/java/org/apache/rocketmq/tools/command/namesrv/GetNamesrvConfigCommandTest.java
index 8d356fd2dbd119852b74958556413d6ab19b8434..8163fd15bc939c1aa1f7a978df5d59d048ad5b60 100644
--- a/tools/src/test/java/org/apache/rocketmq/tools/command/namesrv/GetNamesrvConfigCommandTest.java
+++ b/tools/src/test/java/org/apache/rocketmq/tools/command/namesrv/GetNamesrvConfigCommandTest.java
@@ -81,7 +81,7 @@ public class GetNamesrvConfigCommandTest {
defaultMQAdminExt.shutdown();
}
-// @Ignore
+ // @Ignore
@Test
public void testExecute() throws SubCommandException {
GetNamesrvConfigCommand cmd = new GetNamesrvConfigCommand();