[ISSUE #B5] Reject query assignment requests with missing required body fields instead of failing with NPE

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");

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());