Develop

broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java

@@ -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);
             }
         }
 
@@ -296,7 +296,7 @@ public void run() {
                 public void run() {
                     try {
                         BrokerController.this.protectBroker();
-                    } catch (Exception e) {
+                    } catch (Throwable e) {
                         log.error("protectBroker error.", e);
                     }
                 }
@@ -307,7 +307,7 @@ public void run() {
                 public void run() {
                     try {
                         BrokerController.this.printWaterMark();
-                    } catch (Exception e) {
+                    } catch (Throwable e) {
                         log.error("printWaterMark error.", e);
                     }
                 }

broker/src/main/java/org/apache/rocketmq/broker/client/ClientHousekeepingService.java

@@ -45,8 +45,8 @@ public void start() {
             public void run() {
                 try {
                     ClientHousekeepingService.this.scanExceptionChannel();
-                } catch (Exception e) {
-                    log.error("", e);
+                } catch (Throwable e) {
+                    log.error("Error occurred when scan not active client channels.", e);
                 }
             }
         }, 1000 * 10, 1000 * 10, TimeUnit.MILLISECONDS);

broker/src/main/java/org/apache/rocketmq/broker/processor/PullMessageProcessor.java

@@ -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);
                 }
             }
         };