diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/QueryAssignmentProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/QueryAssignmentProcessor.java index 586e5ec668f..1f18057995e 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/QueryAssignmentProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/QueryAssignmentProcessor.java @@ -104,6 +104,15 @@ private RemotingCommand queryAssignment(ChannelHandlerContext ctx, RemotingComma final RemotingCommand response = RemotingCommand.createResponseCommand(null); final QueryAssignmentResponseBody responseBody = new QueryAssignmentResponseBody(); + // A request without these fields would otherwise fail deep inside the + // lookup tables with an NPE; reject it with a proper error instead. + if (StringUtils.isEmpty(topic) || StringUtils.isEmpty(consumerGroup) || null == messageModel) { + response.setCode(ResponseCode.INVALID_PARAMETER); + response.setRemark("topic, consumerGroup and messageModel can not be null. topic: " + topic + + ", consumerGroup: " + consumerGroup + ", messageModel: " + messageModel); + return response; + } + SetMessageRequestModeRequestBody setMessageRequestModeRequestBody = this.messageRequestModeManager.getMessageRequestMode(topic, consumerGroup); if (setMessageRequestModeRequestBody == null) { @@ -306,6 +315,14 @@ private RemotingCommand setMessageRequestMode(ChannelHandlerContext ctx, final SetMessageRequestModeRequestBody requestBody = SetMessageRequestModeRequestBody.decode(request.getBody(), SetMessageRequestModeRequestBody.class); final String topic = requestBody.getTopic(); + // selectTopicConfig/findSubscriptionGroupConfig would throw an NPE on a + // null key; reject the request with a proper error instead. + if (StringUtils.isEmpty(topic) || StringUtils.isEmpty(requestBody.getConsumerGroup())) { + response.setCode(ResponseCode.INVALID_PARAMETER); + response.setRemark("topic and consumerGroup can not be null. topic: " + topic + + ", consumerGroup: " + requestBody.getConsumerGroup()); + return response; + } if (topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) { response.setCode(ResponseCode.NO_PERMISSION); response.setRemark("retry topic is not allowed to set mode"); diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/QueryAssignmentProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/QueryAssignmentProcessorTest.java index 67ff74897ef..d1c8315b930 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/processor/QueryAssignmentProcessorTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/QueryAssignmentProcessorTest.java @@ -212,12 +212,51 @@ private boolean checkAllocateResult(int popShareQueueNum, int mqSize, int cidSiz return false; } + @Test + public void testQueryAssignmentWithNullTopic() throws Exception { + final RemotingCommand request = createQueryAssignmentRequest(null, group, MessageModel.CLUSTERING); + RemotingCommand responseToReturn = queryAssignmentProcessor.processRequest(handlerContext, request); + assertThat(responseToReturn.getCode()).isEqualTo(ResponseCode.INVALID_PARAMETER); + } + + @Test + public void testQueryAssignmentWithNullConsumerGroup() throws Exception { + final RemotingCommand request = createQueryAssignmentRequest(topic, null, MessageModel.CLUSTERING); + RemotingCommand responseToReturn = queryAssignmentProcessor.processRequest(handlerContext, request); + assertThat(responseToReturn.getCode()).isEqualTo(ResponseCode.INVALID_PARAMETER); + } + + @Test + public void testQueryAssignmentWithNullMessageModel() throws Exception { + final RemotingCommand request = createQueryAssignmentRequest(topic, group, null); + RemotingCommand responseToReturn = queryAssignmentProcessor.processRequest(handlerContext, request); + assertThat(responseToReturn.getCode()).isEqualTo(ResponseCode.INVALID_PARAMETER); + } + + @Test + public void testSetMessageRequestModeWithNullTopic() throws Exception { + final RemotingCommand request = createSetMessageRequestModeRequest(null, group); + RemotingCommand responseToReturn = queryAssignmentProcessor.processRequest(handlerContext, request); + assertThat(responseToReturn.getCode()).isEqualTo(ResponseCode.INVALID_PARAMETER); + } + + @Test + public void testSetMessageRequestModeWithNullConsumerGroup() throws Exception { + final RemotingCommand request = createSetMessageRequestModeRequest(topic, null); + RemotingCommand responseToReturn = queryAssignmentProcessor.processRequest(handlerContext, request); + assertThat(responseToReturn.getCode()).isEqualTo(ResponseCode.INVALID_PARAMETER); + } + private RemotingCommand createQueryAssignmentRequest() { + return createQueryAssignmentRequest(topic, group, MessageModel.CLUSTERING); + } + + private RemotingCommand createQueryAssignmentRequest(String topic, String consumerGroup, MessageModel messageModel) { QueryAssignmentRequestBody requestBody = new QueryAssignmentRequestBody(); requestBody.setTopic(topic); - requestBody.setConsumerGroup(group); + requestBody.setConsumerGroup(consumerGroup); requestBody.setClientId(clientId); - requestBody.setMessageModel(MessageModel.CLUSTERING); + requestBody.setMessageModel(messageModel); requestBody.setStrategyName("AVG"); RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.QUERY_ASSIGNMENT, null); @@ -226,11 +265,15 @@ private RemotingCommand createQueryAssignmentRequest() { } private RemotingCommand createSetMessageRequestModeRequest(String topic) { + return createSetMessageRequestModeRequest(topic, group); + } + + private RemotingCommand createSetMessageRequestModeRequest(String topic, String consumerGroup) { RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.SET_MESSAGE_REQUEST_MODE, null); SetMessageRequestModeRequestBody requestBody = new SetMessageRequestModeRequestBody(); requestBody.setTopic(topic); - requestBody.setConsumerGroup(group); + requestBody.setConsumerGroup(consumerGroup); requestBody.setMode(MessageRequestMode.POP); requestBody.setPopShareQueueNum(0); request.setBody(requestBody.encode());