private RemotingCommand queryConsumerOffset(ChannelHandlerContext ctx, RemotingCommand request)
        throws RemotingCommandException {
        final RemotingCommand response =
            RemotingCommand.createResponseCommand(QueryConsumerOffsetResponseHeader.class);
        final QueryConsumerOffsetResponseHeader responseHeader =
            (QueryConsumerOffsetResponseHeader) response.readCustomHeader();
        final QueryConsumerOffsetRequestHeader requestHeader =
            (QueryConsumerOffsetRequestHeader) request
                .decodeCommandCustomHeader(QueryConsumerOffsetRequestHeader.class);

        TopicQueueMappingContext mappingContext = this.brokerController.getTopicQueueMappingManager().buildTopicQueueMappingContext(requestHeader);
        RemotingCommand rewriteResult = rewriteRequestForStaticTopic(requestHeader, mappingContext);
        if (rewriteResult != null) {
            return rewriteResult;
        }

        long offset =
            this.brokerController.getConsumerOffsetManager().queryOffset(
                requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId());

        if (offset >= 0) {
            responseHeader.setOffset(offset);
            response.setCode(ResponseCode.SUCCESS);
            response.setRemark(null);
        } else {
            long minOffset =
                this.brokerController.getMessageStore().getMinOffsetInQueue(requestHeader.getTopic(),
                    requestHeader.getQueueId());
            if (requestHeader.getSetZeroIfNotFound() != null && Boolean.FALSE.equals(requestHeader.getSetZeroIfNotFound())) {
                response.setCode(ResponseCode.QUERY_NOT_FOUND);
                response.setRemark("Not found, do not set to zero, maybe this group boot first");
            } else if (minOffset <= 0
                && this.brokerController.getMessageStore().checkInMemByConsumeOffset(
                requestHeader.getTopic(), requestHeader.getQueueId(), 0, 1)) {
                responseHeader.setOffset(0L);
                response.setCode(ResponseCode.SUCCESS);
                response.setRemark(null);
            } else {
                response.setCode(ResponseCode.QUERY_NOT_FOUND);
                response.setRemark("Not found, V3_0_6_SNAPSHOT maybe this group consumer boot first");
            }
        }

        RemotingCommand rewriteResponseResult = rewriteResponseForStaticTopic(requestHeader, responseHeader, mappingContext, response.getCode());
        if (rewriteResponseResult != null) {
            return rewriteResponseResult;
        }

        return response;
    }
@Test
    public void testQueryConsumerOffset() throws RemotingCommandException, ExecutionException, InterruptedException {
        RemotingCommand request = buildQueryConsumerOffsetRequest(group, topic, 0, true);
        RemotingCommand response = consumerManageProcessor.processRequest(handlerContext, request);
        assertThat(response).isNotNull();
        assertThat(response.getCode()).isEqualTo(ResponseCode.QUERY_NOT_FOUND);

        when(brokerController.getConsumerOffsetManager()).thenReturn(consumerOffsetManager);
        when(consumerOffsetManager.queryOffset(anyString(),anyString(),anyInt())).thenReturn(0L);
        response = consumerManageProcessor.processRequest(handlerContext, request);
        assertThat(response).isNotNull();
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);

        when(consumerOffsetManager.queryOffset(anyString(),anyString(),anyInt())).thenReturn(-1L);
        when(messageStore.getMinOffsetInQueue(anyString(),anyInt())).thenReturn(-1L);
        when(messageStore.checkInMemByConsumeOffset(anyString(),anyInt(),anyLong(),anyInt())).thenReturn(true);
        response = consumerManageProcessor.processRequest(handlerContext, request);
        assertThat(response).isNotNull();
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);

        TopicQueueMappingManager topicQueueMappingManager = mock(TopicQueueMappingManager.class);
        when(brokerController.getTopicQueueMappingManager()).thenReturn(topicQueueMappingManager);
        when(topicQueueMappingManager.buildTopicQueueMappingContext(any(QueryConsumerOffsetRequestHeader.class))).thenReturn(mappingContext);
        response = consumerManageProcessor.processRequest(handlerContext, request);
        assertThat(response).isNotNull();
        assertThat(response.getCode()).isEqualTo(ResponseCode.NOT_LEADER_FOR_QUEUE);

        List<LogicQueueMappingItem> items = new ArrayList<>();
        LogicQueueMappingItem item1 = createLogicQueueMappingItem("BrokerC", 0, 0L, 0L);
        items.add(item1);
        when(mappingContext.getMappingItemList()).thenReturn(items);
        when(mappingContext.getLeaderItem()).thenReturn(item1);
        when(mappingContext.getCurrentItem()).thenReturn(item1);
        when(mappingContext.isLeader()).thenReturn(true);
        response = consumerManageProcessor.processRequest(handlerContext, request);
        assertThat(response).isNotNull();
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);

        LogicQueueMappingItem item2 = createLogicQueueMappingItem("BrokerA", 0, 0L, 0L);
        items.add(item2);
        QueryConsumerOffsetResponseHeader queryConsumerOffsetResponseHeader = new QueryConsumerOffsetResponseHeader();
        queryConsumerOffsetResponseHeader.setOffset(0L);
        RpcResponse rpcResponse = new RpcResponse(ResponseCode.SUCCESS,queryConsumerOffsetResponseHeader,null);
        when(responseFuture.get()).thenReturn(rpcResponse);
        response = consumerManageProcessor.processRequest(handlerContext, request);
        assertThat(response).isNotNull();
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);

        queryConsumerOffsetResponseHeader.setOffset(-1L);
        rpcResponse = new RpcResponse(ResponseCode.SUCCESS,queryConsumerOffsetResponseHeader,null);
        when(responseFuture.get()).thenReturn(rpcResponse);
        response = consumerManageProcessor.processRequest(handlerContext, request);
        assertThat(response).isNotNull();
        assertThat(response.getCode()).isEqualTo(ResponseCode.QUERY_NOT_FOUND);
    }