Skip to content

Commit

Permalink
Add javadoc to message store.
Browse files Browse the repository at this point in the history
  • Loading branch information
@lizhanhui @dongeforever
lizhanhui authored and dongeforever committed Jun 6, 2017
1 parent 06416b0 commit 482def9
Show file tree
Hide file tree
Showing 9 changed files with 259 additions and 46 deletions.
2 changes: 1 addition & 1 deletion broker/src/main/java/org/apache/rocketmq/broker/client/net/Broker2Client.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,7 @@ public RemotingCommand resetOffset(String topic, String group, long timeStamp, b
long timeStampOffset;
if (timeStamp == -1) {

timeStampOffset = this.brokerController.getMessageStore().getMaxOffsetInQuque(topic, i);
timeStampOffset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, i);
} else {
timeStampOffset = this.brokerController.getMessageStore().getOffsetInQueueByTime(topic, i, timeStamp);
}
Expand Down
4 changes: 2 additions & 2 deletions broker/src/main/java/org/apache/rocketmq/broker/longpolling/PullRequestHoldService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,7 @@ private void checkHoldRequest() {
if (2 == kArray.length) {
String topic = kArray[0];
int queueId = Integer.parseInt(kArray[1]);
final long offset = this.brokerController.getMessageStore().getMaxOffsetInQuque(topic, queueId);
final long offset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId);
try {
this.notifyMessageArriving(topic, queueId, offset);
} catch (Throwable e) {
Expand All @@ -124,7 +124,7 @@ public void notifyMessageArriving(final String topic, final int queueId, final l
for (PullRequest request : requestList) {
long newestOffset = maxOffset;
if (newestOffset <= request.getPullFromThisOffset()) {
newestOffset = this.brokerController.getMessageStore().getMaxOffsetInQuque(topic, queueId);
newestOffset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId);
}

if (newestOffset > request.getPullFromThisOffset()) {
Expand Down
4 changes: 2 additions & 2 deletions broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,7 @@ private boolean offsetBehindMuchThanData(final String topic, ConcurrentHashMap<I

while (it.hasNext() && result) {
Entry<Integer, Long> next = it.next();
long minOffsetInStore = this.brokerController.getMessageStore().getMinOffsetInQuque(topic, next.getKey());
long minOffsetInStore = this.brokerController.getMessageStore().getMinOffsetInQueue(topic, next.getKey());
long offsetInPersist = next.getValue();
result = offsetInPersist <= minOffsetInStore;
}
Expand Down Expand Up @@ -201,7 +201,7 @@ public Map<Integer, Long> queryMinOffsetInAllGroup(final String topic, final Str
String[] topicGroupArr = topicGroup.split(TOPIC_GROUP_SEPARATOR);
if (topic.equals(topicGroupArr[0])) {
for (Entry<Integer, Long> entry : offSetEntry.getValue().entrySet()) {
long minOffset = this.brokerController.getMessageStore().getMinOffsetInQuque(topic, entry.getKey());
long minOffset = this.brokerController.getMessageStore().getMinOffsetInQueue(topic, entry.getKey());
if (entry.getValue() >= minOffset) {
Long offset = queueMinOffset.get(entry.getKey());
if (offset == null) {
Expand Down
20 changes: 10 additions & 10 deletions broker/src/main/java/org/apache/rocketmq/broker/plugin/AbstractPluginMessageStore.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -92,18 +92,18 @@ public GetMessageResult getMessage(String group, String topic, int queueId, long
}

@Override
public long getMaxOffsetInQuque(String topic, int queueId) {
return next.getMaxOffsetInQuque(topic, queueId);
public long getMaxOffsetInQueue(String topic, int queueId) {
return next.getMaxOffsetInQueue(topic, queueId);
}

@Override
public long getMinOffsetInQuque(String topic, int queueId) {
return next.getMinOffsetInQuque(topic, queueId);
public long getMinOffsetInQueue(String topic, int queueId) {
return next.getMinOffsetInQueue(topic, queueId);
}

@Override
public long getCommitLogOffsetInQueue(String topic, int queueId, long cqOffset) {
return next.getCommitLogOffsetInQueue(topic, queueId, cqOffset);
public long getCommitLogOffsetInQueue(String topic, int queueId, long consumeQueueOffset) {
return next.getCommitLogOffsetInQueue(topic, queueId, consumeQueueOffset);
}

@Override
Expand Down Expand Up @@ -152,8 +152,8 @@ public long getEarliestMessageTime(String topic, int queueId) {
}

@Override
public long getMessageStoreTimeStamp(String topic, int queueId, long offset) {
return next.getMessageStoreTimeStamp(topic, queueId, offset);
public long getMessageStoreTimeStamp(String topic, int queueId, long consumeQueueOffset) {
return next.getMessageStoreTimeStamp(topic, queueId, consumeQueueOffset);
}

@Override
Expand All @@ -172,8 +172,8 @@ public boolean appendToCommitLog(long startOffset, byte[] data) {
}

@Override
public void excuteDeleteFilesManualy() {
next.excuteDeleteFilesManualy();
public void executeDeleteFilesManually() {
next.executeDeleteFilesManually();
}

@Override
Expand Down
16 changes: 8 additions & 8 deletions broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -376,7 +376,7 @@ private RemotingCommand getMaxOffset(ChannelHandlerContext ctx, RemotingCommand
final GetMaxOffsetRequestHeader requestHeader =
(GetMaxOffsetRequestHeader) request.decodeCommandCustomHeader(GetMaxOffsetRequestHeader.class);

long offset = this.brokerController.getMessageStore().getMaxOffsetInQuque(requestHeader.getTopic(), requestHeader.getQueueId());
long offset = this.brokerController.getMessageStore().getMaxOffsetInQueue(requestHeader.getTopic(), requestHeader.getQueueId());

responseHeader.setOffset(offset);

Expand All @@ -391,7 +391,7 @@ private RemotingCommand getMinOffset(ChannelHandlerContext ctx, RemotingCommand
final GetMinOffsetRequestHeader requestHeader =
(GetMinOffsetRequestHeader) request.decodeCommandCustomHeader(GetMinOffsetRequestHeader.class);

long offset = this.brokerController.getMessageStore().getMinOffsetInQuque(requestHeader.getTopic(), requestHeader.getQueueId());
long offset = this.brokerController.getMessageStore().getMinOffsetInQueue(requestHeader.getTopic(), requestHeader.getQueueId());

responseHeader.setOffset(offset);
response.setCode(ResponseCode.SUCCESS);
Expand Down Expand Up @@ -537,11 +537,11 @@ private RemotingCommand getTopicStatsInfo(ChannelHandlerContext ctx, RemotingCom
mq.setQueueId(i);

TopicOffset topicOffset = new TopicOffset();
long min = this.brokerController.getMessageStore().getMinOffsetInQuque(topic, i);
long min = this.brokerController.getMessageStore().getMinOffsetInQueue(topic, i);
if (min < 0)
min = 0;

long max = this.brokerController.getMessageStore().getMaxOffsetInQuque(topic, i);
long max = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, i);
if (max < 0)
max = 0;

Expand Down Expand Up @@ -679,7 +679,7 @@ private RemotingCommand getConsumeStats(ChannelHandlerContext ctx, RemotingComma

OffsetWrapper offsetWrapper = new OffsetWrapper();

long brokerOffset = this.brokerController.getMessageStore().getMaxOffsetInQuque(topic, i);
long brokerOffset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, i);
if (brokerOffset < 0)
brokerOffset = 0;

Expand Down Expand Up @@ -862,7 +862,7 @@ private RemotingCommand queryConsumeTimeSpan(ChannelHandlerContext ctx, Remoting
long minTime = this.brokerController.getMessageStore().getEarliestMessageTime(topic, i);
timeSpan.setMinTimeStamp(minTime);

long max = this.brokerController.getMessageStore().getMaxOffsetInQuque(topic, i);
long max = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, i);
long maxTime = this.brokerController.getMessageStore().getMessageStoreTimeStamp(topic, i, max - 1);
timeSpan.setMaxTimeStamp(maxTime);

Expand All @@ -876,7 +876,7 @@ private RemotingCommand queryConsumeTimeSpan(ChannelHandlerContext ctx, Remoting
}
timeSpan.setConsumeTimeStamp(consumeTime);

long maxBrokerOffset = this.brokerController.getMessageStore().getMaxOffsetInQuque(requestHeader.getTopic(), i);
long maxBrokerOffset = this.brokerController.getMessageStore().getMaxOffsetInQueue(requestHeader.getTopic(), i);
if (consumerOffset < maxBrokerOffset) {
long nextTime = this.brokerController.getMessageStore().getMessageStoreTimeStamp(topic, i, consumerOffset);
timeSpan.setDelayTime(System.currentTimeMillis() - nextTime);
Expand Down Expand Up @@ -1126,7 +1126,7 @@ private RemotingCommand fetchAllConsumeStatsInBroker(ChannelHandlerContext ctx,
mq.setBrokerName(this.brokerController.getBrokerConfig().getBrokerName());
mq.setQueueId(i);
OffsetWrapper offsetWrapper = new OffsetWrapper();
long brokerOffset = this.brokerController.getMessageStore().getMaxOffsetInQuque(topic, i);
long brokerOffset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, i);
if (brokerOffset < 0)
brokerOffset = 0;
long consumerOffset = this.brokerController.getConsumerOffsetManager().queryOffset(//
Expand Down
2 changes: 1 addition & 1 deletion broker/src/main/java/org/apache/rocketmq/broker/processor/ConsumerManageProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,7 @@ private RemotingCommand queryConsumerOffset(ChannelHandlerContext ctx, RemotingC
response.setRemark(null);
} else {
long minOffset =
this.brokerController.getMessageStore().getMinOffsetInQuque(requestHeader.getTopic(),
this.brokerController.getMessageStore().getMinOffsetInQueue(requestHeader.getTopic(),
requestHeader.getQueueId());
if (minOffset <= 0
&& !this.brokerController.getMessageStore().checkInDiskByConsumeOffset(
Expand Down
16 changes: 8 additions & 8 deletions store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -580,7 +580,7 @@ public GetMessageResult getMessage(final String group, final String topic, final
/**
*/
public long getMaxOffsetInQuque(String topic, int queueId) {
public long getMaxOffsetInQueue(String topic, int queueId) {
ConsumeQueue logic = this.findConsumeQueue(topic, queueId);
if (logic != null) {
long offset = logic.getMaxOffsetInQueue();
Expand All @@ -593,7 +593,7 @@ public long getMaxOffsetInQuque(String topic, int queueId) {
/**
*/
public long getMinOffsetInQuque(String topic, int queueId) {
public long getMinOffsetInQueue(String topic, int queueId) {
ConsumeQueue logic = this.findConsumeQueue(topic, queueId);
if (logic != null) {
return logic.getMinOffsetInQueue();
Expand All @@ -603,10 +603,10 @@ public long getMinOffsetInQuque(String topic, int queueId) {
}

@Override
public long getCommitLogOffsetInQueue(String topic, int queueId, long cqOffset) {
public long getCommitLogOffsetInQueue(String topic, int queueId, long consumeQueueOffset) {
ConsumeQueue consumeQueue = findConsumeQueue(topic, queueId);
if (consumeQueue != null) {
SelectMappedBufferResult bufferConsumeQueue = consumeQueue.getIndexBuffer(cqOffset);
SelectMappedBufferResult bufferConsumeQueue = consumeQueue.getIndexBuffer(consumeQueueOffset);
if (bufferConsumeQueue != null) {
try {
long offsetPy = bufferConsumeQueue.getByteBuffer().getLong();
Expand Down Expand Up @@ -740,10 +740,10 @@ public long getEarliestMessageTime() {
}

@Override
public long getMessageStoreTimeStamp(String topic, int queueId, long offset) {
public long getMessageStoreTimeStamp(String topic, int queueId, long consumeQueueOffset) {
ConsumeQueue logicQueue = this.findConsumeQueue(topic, queueId);
if (logicQueue != null) {
SelectMappedBufferResult result = logicQueue.getIndexBuffer(offset);
SelectMappedBufferResult result = logicQueue.getIndexBuffer(consumeQueueOffset);
if (result != null) {
try {
final long phyOffset = result.getByteBuffer().getLong();
Expand Down Expand Up @@ -798,7 +798,7 @@ public boolean appendToCommitLog(long startOffset, byte[] data) {
}

@Override
public void excuteDeleteFilesManualy() {
public void executeDeleteFilesManually() {
this.cleanCommitLogService.excuteDeleteFilesManualy();
}

Expand Down Expand Up @@ -1434,7 +1434,7 @@ class CleanCommitLogService {

public void excuteDeleteFilesManualy() {
this.manualDeleteFileSeveralTimes = MAX_MANUAL_DELETE_FILE_TIMES;
DefaultMessageStore.log.info("excuteDeleteFilesManualy was invoked");
DefaultMessageStore.log.info("executeDeleteFilesManually was invoked");
}

public void run() {
Expand Down
Loading

0 comments on commit 482def9

Please sign in to comment.