private RemotingCommand queryCorrectionOffset(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        QueryCorrectionOffsetHeader requestHeader =
            (QueryCorrectionOffsetHeader) request.decodeCommandCustomHeader(QueryCorrectionOffsetHeader.class);

        Map<Integer, Long> correctionOffset = this.brokerController.getConsumerOffsetManager()
            .queryMinOffsetInAllGroup(requestHeader.getTopic(), requestHeader.getFilterGroups());

        Map<Integer, Long> compareOffset =
            this.brokerController.getConsumerOffsetManager().queryOffset(requestHeader.getCompareGroup(), requestHeader.getTopic());

        if (compareOffset != null && !compareOffset.isEmpty()) {
            for (Map.Entry<Integer, Long> entry : compareOffset.entrySet()) {
                Integer queueId = entry.getKey();
                correctionOffset.put(queueId,
                    correctionOffset.get(queueId) > entry.getValue() ? Long.MAX_VALUE : correctionOffset.get(queueId));
            }
        }

        QueryCorrectionOffsetBody body = new QueryCorrectionOffsetBody();
        body.setCorrectionOffsets(correctionOffset);
        response.setBody(body.encode());
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }
@Test
    public void testQueryCorrectionOffset() throws RemotingCommandException {
        Map<Integer, Long> correctionOffsetMap = new HashMap<>();
        correctionOffsetMap.put(0, 100L);
        correctionOffsetMap.put(1, 200L);
        Map<Integer, Long> compareOffsetMap = new HashMap<>();
        compareOffsetMap.put(0, 80L);
        compareOffsetMap.put(1, 300L);
        when(brokerController.getConsumerOffsetManager()).thenReturn(consumerOffsetManager);
        when(consumerOffsetManager.queryMinOffsetInAllGroup(anyString(),anyString())).thenReturn(correctionOffsetMap);
        when(consumerOffsetManager.queryOffset(anyString(),anyString())).thenReturn(compareOffsetMap);
        QueryCorrectionOffsetHeader queryCorrectionOffsetHeader = new QueryCorrectionOffsetHeader();
        queryCorrectionOffsetHeader.setTopic("topic");
        queryCorrectionOffsetHeader.setCompareGroup("group");
        queryCorrectionOffsetHeader.setFilterGroups("");
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.QUERY_CORRECTION_OFFSET, queryCorrectionOffsetHeader);
        request.makeCustomHeaderToNet();
        RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
        QueryCorrectionOffsetBody body = RemotingSerializable.decode(response.getBody(), QueryCorrectionOffsetBody.class);
        Map<Integer, Long> correctionOffsets = body.getCorrectionOffsets();
        assertThat(correctionOffsets.get(0)).isEqualTo(Long.MAX_VALUE);
        assertThat(correctionOffsets.get(1)).isEqualTo(200L);
    }