From 9e0fb1b0293aebbe5019e55c0fdbd356611b3d16 Mon Sep 17 00:00:00 2001 From: rongtong Date: Wed, 8 Mar 2023 17:23:06 +0800 Subject: [PATCH] [ISSUE #6169] Fix NPE when timerWheel disabled (#6184) * Fix some NPE when timerWheel disabled * Add UT --- .../broker/processor/PopMessageProcessor.java | 13 ++- .../broker/processor/PopReviveService.java | 5 + .../processor/PopMessageProcessorTest.java | 16 ++- .../metrics/DefaultStoreMetricsManager.java | 102 +++++++++--------- 4 files changed, 82 insertions(+), 54 deletions(-) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java index cd45953269..d63fbe6212 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java @@ -278,19 +278,26 @@ public class PopMessageProcessor implements NettyRequestProcessor { if (requestHeader.isTimeoutTooMuch()) { response.setCode(ResponseCode.POLLING_TIMEOUT); - response.setRemark(String.format("the broker[%s] poping message is timeout too much", + response.setRemark(String.format("the broker[%s] pop message is timeout too much", this.brokerController.getBrokerConfig().getBrokerIP1())); return response; } if (!PermName.isReadable(this.brokerController.getBrokerConfig().getBrokerPermission())) { response.setCode(ResponseCode.NO_PERMISSION); - response.setRemark(String.format("the broker[%s] poping message is forbidden", + response.setRemark(String.format("the broker[%s] pop message is forbidden", this.brokerController.getBrokerConfig().getBrokerIP1())); return response; } if (requestHeader.getMaxMsgNums() > 32) { response.setCode(ResponseCode.SYSTEM_ERROR); - response.setRemark(String.format("the broker[%s] poping message's num is greater than 32", + response.setRemark(String.format("the broker[%s] pop message's num is greater than 32", + this.brokerController.getBrokerConfig().getBrokerIP1())); + return response; + } + + if (!brokerController.getMessageStore().getMessageStoreConfig().isTimerWheelEnable()) { + response.setCode(ResponseCode.SYSTEM_ERROR); + response.setRemark(String.format("the broker[%s] pop message is forbidden because timerWheelEnable is false", this.brokerController.getBrokerConfig().getBrokerIP1())); return response; } diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java index a63cd2930b..116cb0f82d 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java @@ -600,6 +600,11 @@ public class PopReviveService extends ServiceThread { continue; } + if (!brokerController.getMessageStore().getMessageStoreConfig().isTimerWheelEnable()) { + POP_LOGGER.warn("skip revive topic because timerWheelEnable is false"); + continue; + } + POP_LOGGER.info("start revive topic={}, reviveQueueId={}", reviveTopic, queueId); ConsumeReviveObj consumeReviveObj = new ConsumeReviveObj(); consumeReviveMessage(consumeReviveObj); diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/PopMessageProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/PopMessageProcessorTest.java index aabc682204..44f04066ca 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/processor/PopMessageProcessorTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/PopMessageProcessorTest.java @@ -91,6 +91,7 @@ public class PopMessageProcessorTest { @Test public void testProcessRequest_TopicNotExist() throws RemotingCommandException { + when(messageStore.getMessageStoreConfig()).thenReturn(new MessageStoreConfig()); brokerController.getTopicConfigManager().getTopicConfigTable().remove(topic); final RemotingCommand request = createPopMsgCommand(); RemotingCommand response = popMessageProcessor.processRequest(handlerContext, request); @@ -102,6 +103,7 @@ public class PopMessageProcessorTest { @Test public void testProcessRequest_Found() throws RemotingCommandException, InterruptedException { GetMessageResult getMessageResult = createGetMessageResult(1); + when(messageStore.getMessageStoreConfig()).thenReturn(new MessageStoreConfig()); when(messageStore.getMessageAsync(anyString(), anyString(), anyInt(), anyLong(), anyInt(), any())).thenReturn(CompletableFuture.completedFuture(getMessageResult)); final RemotingCommand request = createPopMsgCommand(); @@ -115,6 +117,7 @@ public class PopMessageProcessorTest { public void testProcessRequest_MsgWasRemoving() throws RemotingCommandException { GetMessageResult getMessageResult = createGetMessageResult(1); getMessageResult.setStatus(GetMessageStatus.MESSAGE_WAS_REMOVING); + when(messageStore.getMessageStoreConfig()).thenReturn(new MessageStoreConfig()); when(messageStore.getMessageAsync(anyString(), anyString(), anyInt(), anyLong(), anyInt(), any())).thenReturn(CompletableFuture.completedFuture(getMessageResult)); final RemotingCommand request = createPopMsgCommand(); @@ -128,6 +131,7 @@ public class PopMessageProcessorTest { public void testProcessRequest_NoMsgInQueue() throws RemotingCommandException { GetMessageResult getMessageResult = createGetMessageResult(0); getMessageResult.setStatus(GetMessageStatus.NO_MESSAGE_IN_QUEUE); + when(messageStore.getMessageStoreConfig()).thenReturn(new MessageStoreConfig()); when(messageStore.getMessageAsync(anyString(), anyString(), anyInt(), anyLong(), anyInt(), any())).thenReturn(CompletableFuture.completedFuture(getMessageResult)); final RemotingCommand request = createPopMsgCommand(); @@ -135,6 +139,17 @@ public class PopMessageProcessorTest { assertThat(response).isNull(); } + @Test + public void testProcessRequest_whenTimerWheelIsFalse() throws RemotingCommandException { + MessageStoreConfig messageStoreConfig = new MessageStoreConfig(); + messageStoreConfig.setTimerWheelEnable(false); + when(messageStore.getMessageStoreConfig()).thenReturn(messageStoreConfig); + final RemotingCommand request = createPopMsgCommand(); + RemotingCommand response = popMessageProcessor.processRequest(handlerContext, request); + assertThat(response).isNotNull(); + assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR); + assertThat(response.getRemark()).contains("pop message is forbidden because timerWheelEnable is false"); + } private RemotingCommand createPopMsgCommand() { PopMessageRequestHeader requestHeader = new PopMessageRequestHeader(); @@ -152,7 +167,6 @@ public class PopMessageProcessorTest { return request; } - private GetMessageResult createGetMessageResult(int msgCnt) { GetMessageResult getMessageResult = new GetMessageResult(); getMessageResult.setStatus(GetMessageStatus.FOUND); diff --git a/store/src/main/java/org/apache/rocketmq/store/metrics/DefaultStoreMetricsManager.java b/store/src/main/java/org/apache/rocketmq/store/metrics/DefaultStoreMetricsManager.java index 6862652926..9132761a66 100644 --- a/store/src/main/java/org/apache/rocketmq/store/metrics/DefaultStoreMetricsManager.java +++ b/store/src/main/java/org/apache/rocketmq/store/metrics/DefaultStoreMetricsManager.java @@ -116,57 +116,59 @@ public class DefaultStoreMetricsManager { measurement.record(System.currentTimeMillis() - earliestMessageTime, newAttributesBuilder().build()); }); - timerEnqueueLag = meter.gaugeBuilder(GAUGE_TIMER_ENQUEUE_LAG) - .setDescription("Timer enqueue messages lag") - .ofLongs() - .buildWithCallback(measurement -> { - TimerMessageStore timerMessageStore = messageStore.getTimerMessageStore(); - measurement.record(timerMessageStore.getEnqueueBehindMessages(), newAttributesBuilder().build()); - }); + if (messageStore.getMessageStoreConfig().isTimerWheelEnable()) { + timerEnqueueLag = meter.gaugeBuilder(GAUGE_TIMER_ENQUEUE_LAG) + .setDescription("Timer enqueue messages lag") + .ofLongs() + .buildWithCallback(measurement -> { + TimerMessageStore timerMessageStore = messageStore.getTimerMessageStore(); + measurement.record(timerMessageStore.getEnqueueBehindMessages(), newAttributesBuilder().build()); + }); - timerEnqueueLatency = meter.gaugeBuilder(GAUGE_TIMER_ENQUEUE_LATENCY) - .setDescription("Timer enqueue latency") - .setUnit("milliseconds") - .ofLongs() - .buildWithCallback(measurement -> { - TimerMessageStore timerMessageStore = messageStore.getTimerMessageStore(); - measurement.record(timerMessageStore.getEnqueueBehindMillis(), newAttributesBuilder().build()); - }); - timerDequeueLag = meter.gaugeBuilder(GAUGE_TIMER_DEQUEUE_LAG) - .setDescription("Timer dequeue messages lag") - .ofLongs() - .buildWithCallback(measurement -> { - TimerMessageStore timerMessageStore = messageStore.getTimerMessageStore(); - measurement.record(timerMessageStore.getDequeueBehindMessages(), newAttributesBuilder().build()); - }); - timerDequeueLatency = meter.gaugeBuilder(GAUGE_TIMER_DEQUEUE_LATENCY) - .setDescription("Timer dequeue latency") - .setUnit("milliseconds") - .ofLongs() - .buildWithCallback(measurement -> { - TimerMessageStore timerMessageStore = messageStore.getTimerMessageStore(); - measurement.record(timerMessageStore.getDequeueBehind(), newAttributesBuilder().build()); - }); - timingMessages = meter.gaugeBuilder(GAUGE_TIMING_MESSAGES) - .setDescription("Current message number in timing") - .ofLongs() - .buildWithCallback(measurement -> { - TimerMessageStore timerMessageStore = messageStore.getTimerMessageStore(); - timerMessageStore.getTimerMetrics() - .getTimingCount() - .forEach((topic, metric) -> { - measurement.record( - metric.getCount().get(), - newAttributesBuilder().put(LABEL_TOPIC, topic).build() - ); - }); - }); - timerDequeueTotal = meter.counterBuilder(COUNTER_TIMER_DEQUEUE_TOTAL) - .setDescription("Total number of timer dequeue") - .build(); - timerEnqueueTotal = meter.counterBuilder(COUNTER_TIMER_ENQUEUE_TOTAL) - .setDescription("Total number of timer enqueue") - .build(); + timerEnqueueLatency = meter.gaugeBuilder(GAUGE_TIMER_ENQUEUE_LATENCY) + .setDescription("Timer enqueue latency") + .setUnit("milliseconds") + .ofLongs() + .buildWithCallback(measurement -> { + TimerMessageStore timerMessageStore = messageStore.getTimerMessageStore(); + measurement.record(timerMessageStore.getEnqueueBehindMillis(), newAttributesBuilder().build()); + }); + timerDequeueLag = meter.gaugeBuilder(GAUGE_TIMER_DEQUEUE_LAG) + .setDescription("Timer dequeue messages lag") + .ofLongs() + .buildWithCallback(measurement -> { + TimerMessageStore timerMessageStore = messageStore.getTimerMessageStore(); + measurement.record(timerMessageStore.getDequeueBehindMessages(), newAttributesBuilder().build()); + }); + timerDequeueLatency = meter.gaugeBuilder(GAUGE_TIMER_DEQUEUE_LATENCY) + .setDescription("Timer dequeue latency") + .setUnit("milliseconds") + .ofLongs() + .buildWithCallback(measurement -> { + TimerMessageStore timerMessageStore = messageStore.getTimerMessageStore(); + measurement.record(timerMessageStore.getDequeueBehind(), newAttributesBuilder().build()); + }); + timingMessages = meter.gaugeBuilder(GAUGE_TIMING_MESSAGES) + .setDescription("Current message number in timing") + .ofLongs() + .buildWithCallback(measurement -> { + TimerMessageStore timerMessageStore = messageStore.getTimerMessageStore(); + timerMessageStore.getTimerMetrics() + .getTimingCount() + .forEach((topic, metric) -> { + measurement.record( + metric.getCount().get(), + newAttributesBuilder().put(LABEL_TOPIC, topic).build() + ); + }); + }); + timerDequeueTotal = meter.counterBuilder(COUNTER_TIMER_DEQUEUE_TOTAL) + .setDescription("Total number of timer dequeue") + .build(); + timerEnqueueTotal = meter.counterBuilder(COUNTER_TIMER_ENQUEUE_TOTAL) + .setDescription("Total number of timer enqueue") + .build(); + } } public static void incTimerDequeueCount(String topic) {