Skip to content

Commit

Permalink
ROCKETMQ-6: Use logger for exceptions instead of e.printStackTrace().
Browse files Browse the repository at this point in the history
Signed-off-by: shroman <[email protected]>
  • Loading branch information
shroman committed Jul 5, 2017
1 parent 9ad9ad0 commit 246be9e
Showing 28 changed files with 227 additions and 188 deletions.
Original file line number Diff line number Diff line change
@@ -210,7 +210,7 @@ public boolean initialize() throws CloneNotSupportedException {
this.messageStore.getDispatcherList().addFirst(new CommitLogDispatcherCalcBitMap(this.brokerConfig, this.consumerFilterManager));
} catch (IOException e) {
result = false;
e.printStackTrace();
log.error("Failed to initialize", e);
}
}

Original file line number Diff line number Diff line change
@@ -67,7 +67,7 @@
import org.slf4j.LoggerFactory;

public class PullMessageProcessor implements NettyRequestProcessor {
private static final Logger LOG = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
private static final Logger log = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
private final BrokerController brokerController;
private List<ConsumeMessageHook> consumeMessageHookList;

@@ -94,9 +94,7 @@ private RemotingCommand processRequest(final Channel channel, RemotingCommand re

response.setOpaque(request.getOpaque());

if (LOG.isDebugEnabled()) {
LOG.debug("receive PullMessage request command, {}", request);
}
log.debug("receive PullMessage request command, {}", request);

if (!PermName.isReadable(this.brokerController.getBrokerConfig().getBrokerPermission())) {
response.setCode(ResponseCode.NO_PERMISSION);
@@ -126,7 +124,7 @@ private RemotingCommand processRequest(final Channel channel, RemotingCommand re

TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
if (null == topicConfig) {
LOG.error("The topic {} not exist, consumer: {} ", requestHeader.getTopic(), RemotingHelper.parseChannelRemoteAddr(channel));
log.error("the topic {} not exist, consumer: {}", requestHeader.getTopic(), RemotingHelper.parseChannelRemoteAddr(channel));
response.setCode(ResponseCode.TOPIC_NOT_EXIST);
response.setRemark(String.format("topic[%s] not exist, apply first please! %s", requestHeader.getTopic(), FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL)));
return response;
@@ -141,7 +139,7 @@ private RemotingCommand processRequest(final Channel channel, RemotingCommand re
if (requestHeader.getQueueId() < 0 || requestHeader.getQueueId() >= topicConfig.getReadQueueNums()) {
String errorInfo = String.format("queueId[%d] is illegal, topic:[%s] topicConfig.readQueueNums:[%d] consumer:[%s]",
requestHeader.getQueueId(), requestHeader.getTopic(), topicConfig.getReadQueueNums(), channel.remoteAddress());
LOG.warn(errorInfo);
log.warn(errorInfo);
response.setCode(ResponseCode.SYSTEM_ERROR);
response.setRemark(errorInfo);
return response;
@@ -162,7 +160,7 @@ private RemotingCommand processRequest(final Channel channel, RemotingCommand re
assert consumerFilterData != null;
}
} catch (Exception e) {
LOG.warn("Parse the consumer's subscription[{}] failed, group: {}", requestHeader.getSubscription(), //
log.warn("Parse the consumer's subscription[{}] failed, group: {}", requestHeader.getSubscription(), //
requestHeader.getConsumerGroup());
response.setCode(ResponseCode.SUBSCRIPTION_PARSE_FAILED);
response.setRemark("parse the consumer's subscription failed");
@@ -172,7 +170,7 @@ private RemotingCommand processRequest(final Channel channel, RemotingCommand re
ConsumerGroupInfo consumerGroupInfo =
this.brokerController.getConsumerManager().getConsumerGroupInfo(requestHeader.getConsumerGroup());
if (null == consumerGroupInfo) {
LOG.warn("The consumer's group info not exist, group: {}", requestHeader.getConsumerGroup());
log.warn("the consumer's group info not exist, group: {}", requestHeader.getConsumerGroup());
response.setCode(ResponseCode.SUBSCRIPTION_NOT_EXIST);
response.setRemark("the consumer's group info not exist" + FAQUrl.suggestTodo(FAQUrl.SAME_GROUP_DIFFERENT_TOPIC));
return response;
@@ -187,14 +185,14 @@ private RemotingCommand processRequest(final Channel channel, RemotingCommand re

subscriptionData = consumerGroupInfo.findSubscriptionData(requestHeader.getTopic());
if (null == subscriptionData) {
LOG.warn("The consumer's subscription not exist, group: {}, topic:{}", requestHeader.getConsumerGroup(), requestHeader.getTopic());
log.warn("the consumer's subscription not exist, group: {}, topic:{}", requestHeader.getConsumerGroup(), requestHeader.getTopic());
response.setCode(ResponseCode.SUBSCRIPTION_NOT_EXIST);
response.setRemark("the consumer's subscription not exist" + FAQUrl.suggestTodo(FAQUrl.SAME_GROUP_DIFFERENT_TOPIC));
return response;
}

if (subscriptionData.getSubVersion() < requestHeader.getSubVersion()) {
LOG.warn("The broker's subscription is not latest, group: {} {}", requestHeader.getConsumerGroup(),
log.warn("The broker's subscription is not latest, group: {} {}", requestHeader.getConsumerGroup(),
subscriptionData.getSubString());
response.setCode(ResponseCode.SUBSCRIPTION_NOT_LATEST);
response.setRemark("the consumer's subscription not latest");
@@ -209,7 +207,7 @@ private RemotingCommand processRequest(final Channel channel, RemotingCommand re
return response;
}
if (consumerFilterData.getClientVersion() < requestHeader.getSubVersion()) {
LOG.warn("The broker's consumer filter data is not latest, group: {}, topic: {}, serverV: {}, clientV: {}",
log.warn("The broker's consumer filter data is not latest, group: {}, topic: {}, serverV: {}, clientV: {}",
requestHeader.getConsumerGroup(), requestHeader.getTopic(), consumerFilterData.getClientVersion(), requestHeader.getSubVersion());
response.setCode(ResponseCode.FILTER_DATA_NOT_LATEST);
response.setRemark("the consumer's consumer filter data not latest");
@@ -287,12 +285,12 @@ private RemotingCommand processRequest(final Channel channel, RemotingCommand re
response.setCode(ResponseCode.PULL_OFFSET_MOVED);

// XXX: warn and notify me
LOG.info("the broker store no queue data, fix the request offset {} to {}, Topic: {} QueueId: {} Consumer Group: {}", //
requestHeader.getQueueOffset(), //
getMessageResult.getNextBeginOffset(), //
requestHeader.getTopic(), //
requestHeader.getQueueId(), //
requestHeader.getConsumerGroup()//
log.info("the broker store no queue data, fix the request offset {} to {}, Topic: {} QueueId: {} Consumer Group: {}", //
requestHeader.getQueueOffset(), //
getMessageResult.getNextBeginOffset(), //
requestHeader.getTopic(), //
requestHeader.getQueueId(), //
requestHeader.getConsumerGroup()//
);
} else {
response.setCode(ResponseCode.PULL_NOT_FOUND);
@@ -307,16 +305,17 @@ private RemotingCommand processRequest(final Channel channel, RemotingCommand re
case OFFSET_OVERFLOW_BADLY:
response.setCode(ResponseCode.PULL_OFFSET_MOVED);
// XXX: warn and notify me
LOG.info("The request offset:{} over flow badly, broker max offset:{} , consumer: {}", requestHeader.getQueueOffset(), getMessageResult.getMaxOffset(), channel.remoteAddress());
log.info("the request offset: {} over flow badly, broker max offset: {}, consumer: {}",
requestHeader.getQueueOffset(), getMessageResult.getMaxOffset(), channel.remoteAddress());
break;
case OFFSET_OVERFLOW_ONE:
response.setCode(ResponseCode.PULL_NOT_FOUND);
break;
case OFFSET_TOO_SMALL:
response.setCode(ResponseCode.PULL_OFFSET_MOVED);
LOG.info("The request offset is too small. group={}, topic={}, requestOffset={}, brokerMinOffset={}, clientIp={}",
requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueOffset(),
getMessageResult.getMinOffset(), channel.remoteAddress());
log.info("the request offset too small. group={}, topic={}, requestOffset={}, brokerMinOffset={}, clientIp={}",
requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueOffset(),
getMessageResult.getMinOffset(), channel.remoteAddress());
break;
default:
assert false;
@@ -391,12 +390,12 @@ private RemotingCommand processRequest(final Channel channel, RemotingCommand re
public void operationComplete(ChannelFuture future) throws Exception {
getMessageResult.release();
if (!future.isSuccess()) {
LOG.error("Fail to transfer messages from page cache to {}", channel.remoteAddress(), future.cause());
log.error("transfer many message by pagecache failed, {}", channel.remoteAddress(), future.cause());
}
}
});
} catch (Throwable e) {
LOG.error("Error occurred when transferring messages from page cache", e);
log.error("transfer many message by pagecache exception", e);
getMessageResult.release();
}

@@ -437,16 +436,16 @@ public void operationComplete(ChannelFuture future) throws Exception {
event.setOffsetRequest(requestHeader.getQueueOffset());
event.setOffsetNew(getMessageResult.getNextBeginOffset());
this.generateOffsetMovedEvent(event);
LOG.warn(
"PULL_OFFSET_MOVED:correction offset. topic={}, groupId={}, requestOffset={}, newOffset={}, suggestBrokerId={}",
requestHeader.getTopic(), requestHeader.getConsumerGroup(), event.getOffsetRequest(), event.getOffsetNew(),
responseHeader.getSuggestWhichBrokerId());
log.warn(
"PULL_OFFSET_MOVED:correction offset. topic={}, groupId={}, requestOffset={}, newOffset={}, suggestBrokerId={}",
requestHeader.getTopic(), requestHeader.getConsumerGroup(), event.getOffsetRequest(), event.getOffsetNew(),
responseHeader.getSuggestWhichBrokerId());
} else {
responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getBrokerId());
response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
LOG.warn("PULL_OFFSET_MOVED:none correction. topic={}, groupId={}, requestOffset={}, suggestBrokerId={}",
requestHeader.getTopic(), requestHeader.getConsumerGroup(), requestHeader.getQueueOffset(),
responseHeader.getSuggestWhichBrokerId());
log.warn("PULL_OFFSET_MOVED:none correction. topic={}, groupId={}, requestOffset={}, suggestBrokerId={}",
requestHeader.getTopic(), requestHeader.getConsumerGroup(), requestHeader.getQueueOffset(),
responseHeader.getSuggestWhichBrokerId());
}

break;
@@ -525,7 +524,7 @@ private void generateOffsetMovedEvent(final OffsetMovedEvent event) {

PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
} catch (Exception e) {
LOG.warn(String.format("GenerateOffsetMovedEvent Exception, %s", event.toString()), e);
log.warn(String.format("generateOffsetMovedEvent Exception, %s", event.toString()), e);
}
}

@@ -544,20 +543,21 @@ public void run() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (!future.isSuccess()) {
LOG.error("ProcessRequestWrapper response to {} failed", future.channel().remoteAddress(), future.cause());
LOG.error(request.toString());
LOG.error(response.toString());
log.error("processRequestWrapper response to {} failed",
future.channel().remoteAddress(), future.cause());
log.error(request.toString());
log.error(response.toString());
}
}
});
} catch (Throwable e) {
LOG.error("ProcessRequestWrapper process request over, but response failed", e);
LOG.error(request.toString());
LOG.error(response.toString());
log.error("processRequestWrapper process request over, but response failed", e);
log.error(request.toString());
log.error(response.toString());
}
}
} catch (RemotingCommandException e1) {
LOG.error("ExecuteRequestWhenWakeup run", e1);
log.error("excuteRequestWhenWakeup run", e1);
}
}
};
Loading
Oops, something went wrong.

0 comments on commit 246be9e

Please sign in to comment.