private RemotingCommand updateConsumerOffset(ChannelHandlerContext ctx, RemotingCommand request)
        throws RemotingCommandException {

        final RemotingCommand response =
            RemotingCommand.createResponseCommand(UpdateConsumerOffsetResponseHeader.class);

        final UpdateConsumerOffsetRequestHeader requestHeader =
            (UpdateConsumerOffsetRequestHeader)
                request.decodeCommandCustomHeader(UpdateConsumerOffsetRequestHeader.class);

        TopicQueueMappingContext mappingContext =
            this.brokerController.getTopicQueueMappingManager().buildTopicQueueMappingContext(requestHeader);

        RemotingCommand rewriteResult = rewriteRequestForStaticTopic(requestHeader, mappingContext);
        if (rewriteResult != null) {
            return rewriteResult;
        }

        String topic = requestHeader.getTopic();
        String group = requestHeader.getConsumerGroup();
        Integer queueId = requestHeader.getQueueId();
        Long offset = requestHeader.getCommitOffset();

        if (!this.brokerController.getSubscriptionGroupManager().containsSubscriptionGroup(group)) {
            response.setCode(ResponseCode.SUBSCRIPTION_GROUP_NOT_EXIST);
            response.setRemark("Group " + group + " not exist!");
            return response;
        }

        if (!this.brokerController.getTopicConfigManager().containsTopic(requestHeader.getTopic())) {
            response.setCode(ResponseCode.TOPIC_NOT_EXIST);
            response.setRemark("Topic " + topic + " not exist!");
            return response;
        }

        if (queueId == null) {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("QueueId is null, topic is " + topic);
            return response;
        }

        if (offset == null) {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("Offset is null, topic is " + topic);
            return response;
        }

        ConsumerOffsetManager consumerOffsetManager = brokerController.getConsumerOffsetManager();
        if (this.brokerController.getBrokerConfig().isUseServerSideResetOffset()) {
            // Note, ignoring this update offset request
            if (consumerOffsetManager.hasOffsetReset(topic, group, queueId)) {
                response.setCode(ResponseCode.SUCCESS);
                response.setRemark("Offset has been previously reset");
                LOGGER.info("Update consumer offset is rejected because of previous offset-reset. Group={}, " +
                    "Topic={}, QueueId={}, Offset={}", group, topic, queueId, offset);
                return response;
            }
        }

        this.brokerController.getConsumerOffsetManager().commitOffset(
            RemotingHelper.parseChannelRemoteAddr(ctx.channel()), group, topic, queueId, offset);
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }
@Test
    public void testUpdateConsumerOffset() throws RemotingCommandException {
        when(brokerController.getConsumerOffsetManager()).thenReturn(consumerOffsetManager);
        when(consumerOffsetManager.hasOffsetReset(anyString(),anyString(),anyInt())).thenReturn(true);
        RemotingCommand request = buildUpdateConsumerOffsetRequest(group, topic, 0, 0);
        RemotingCommand response = consumerManageProcessor.processRequest(handlerContext, request);
        assertThat(response).isNotNull();
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);

        when(consumerOffsetManager.hasOffsetReset(anyString(),anyString(),anyInt())).thenReturn(false);
        response = consumerManageProcessor.processRequest(handlerContext, request);
        assertThat(response).isNotNull();
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
    }