public RemotingCommand rewriteRequestForStaticTopic(final UpdateConsumerOffsetRequestHeader requestHeader,
        final TopicQueueMappingContext mappingContext) {
        try {
            if (mappingContext.getMappingDetail() == null) {
                return null;
            }
            TopicQueueMappingDetail mappingDetail = mappingContext.getMappingDetail();
            if (!mappingContext.isLeader()) {
                return buildErrorResponse(ResponseCode.NOT_LEADER_FOR_QUEUE, String.format("%s-%d does not exit in request process of current broker %s", requestHeader.getTopic(), requestHeader.getQueueId(), mappingDetail.getBname()));
            }
            Long globalOffset = requestHeader.getCommitOffset();
            LogicQueueMappingItem mappingItem = TopicQueueMappingUtils.findLogicQueueMappingItem(mappingContext.getMappingItemList(), globalOffset, true);
            requestHeader.setQueueId(mappingItem.getQueueId());
            requestHeader.setLo(false);
            requestHeader.setBrokerName(mappingItem.getBname());
            requestHeader.setCommitOffset(mappingItem.computePhysicalQueueOffset(globalOffset));
            //leader, let it go, do not need to rewrite the response
            if (mappingDetail.getBname().equals(mappingItem.getBname())) {
                return null;
            }
            RpcRequest rpcRequest = new RpcRequest(RequestCode.UPDATE_CONSUMER_OFFSET, requestHeader, null);
            RpcResponse rpcResponse = this.brokerController.getBrokerOuterAPI().getRpcClient().invoke(rpcRequest, this.brokerController.getBrokerConfig().getForwardTimeout()).get();
            if (rpcResponse.getException() != null) {
                throw rpcResponse.getException();
            }
            return RpcClientUtils.createCommandForRpcResponse(rpcResponse);
        } catch (Throwable t) {
            return buildErrorResponse(ResponseCode.SYSTEM_ERROR, t.getMessage());
        }
    }
@Test
    public void testRewriteRequestForStaticTopic() throws RpcException, ExecutionException, InterruptedException {
        UpdateConsumerOffsetRequestHeader requestHeader = new UpdateConsumerOffsetRequestHeader();
        requestHeader.setConsumerGroup(group);
        requestHeader.setTopic(topic);
        requestHeader.setQueueId(0);
        requestHeader.setCommitOffset(0L);

        RemotingCommand response = consumerManageProcessor.rewriteRequestForStaticTopic(requestHeader, mappingContext);
        assertThat(response).isNotNull();
        assertThat(response.getCode()).isEqualTo(ResponseCode.NOT_LEADER_FOR_QUEUE);

        List<LogicQueueMappingItem> items = new ArrayList<>();
        LogicQueueMappingItem item = createLogicQueueMappingItem("BrokerC", 0, 0L, 0L);
        items.add(item);
        when(mappingContext.getMappingItemList()).thenReturn(items);
        when(mappingContext.isLeader()).thenReturn(true);
        RpcResponse rpcResponse = new RpcResponse(ResponseCode.SUCCESS,new UpdateConsumerOffsetResponseHeader(),null);
        when(responseFuture.get()).thenReturn(rpcResponse);
        response = consumerManageProcessor.rewriteRequestForStaticTopic(requestHeader, mappingContext);
        assertThat(response).isNotNull();
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
    }