public RemotingCommand resetOffset(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        final ResetOffsetRequestHeader requestHeader =
            (ResetOffsetRequestHeader) request.decodeCommandCustomHeader(ResetOffsetRequestHeader.class);
        logger.info("invoke reset offset operation from broker. brokerAddr={}, topic={}, group={}, timestamp={}",
            RemotingHelper.parseChannelRemoteAddr(ctx.channel()), requestHeader.getTopic(), requestHeader.getGroup(),
            requestHeader.getTimestamp());
        Map<MessageQueue, Long> offsetTable = new HashMap<>();
        if (request.getBody() != null) {
            ResetOffsetBody body = ResetOffsetBody.decode(request.getBody(), ResetOffsetBody.class);
            offsetTable = body.getOffsetTable();
        }
        this.mqClientFactory.resetOffset(requestHeader.getTopic(), requestHeader.getGroup(), offsetTable);
        return null;
    }
@Test
    public void testResetOffset() throws Exception {
        ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
        RemotingCommand request = mock(RemotingCommand.class);
        when(request.getCode()).thenReturn(RequestCode.RESET_CONSUMER_CLIENT_OFFSET);
        ResetOffsetBody offsetBody = new ResetOffsetBody();
        when(request.getBody()).thenReturn(RemotingSerializable.encode(offsetBody));
        ResetOffsetRequestHeader requestHeader = new ResetOffsetRequestHeader();
        when(request.decodeCommandCustomHeader(ResetOffsetRequestHeader.class)).thenReturn(requestHeader);
        assertNull(processor.processRequest(ctx, request));
    }