public synchronized void resetOffset(String topic, String group, Map<MessageQueue, Long> offsetTable) {
        DefaultMQPushConsumerImpl consumer = null;
        try {
            MQConsumerInner impl = this.consumerTable.get(group);
            if (impl instanceof DefaultMQPushConsumerImpl) {
                consumer = (DefaultMQPushConsumerImpl) impl;
            } else {
                log.info("[reset-offset] consumer dose not exist. group={}", group);
                return;
            }
            consumer.suspend();

            ConcurrentMap<MessageQueue, ProcessQueue> processQueueTable = consumer.getRebalanceImpl().getProcessQueueTable();
            for (Map.Entry<MessageQueue, ProcessQueue> entry : processQueueTable.entrySet()) {
                MessageQueue mq = entry.getKey();
                if (topic.equals(mq.getTopic()) && offsetTable.containsKey(mq)) {
                    ProcessQueue pq = entry.getValue();
                    pq.setDropped(true);
                    pq.clear();
                }
            }

            try {
                TimeUnit.SECONDS.sleep(10);
            } catch (InterruptedException ignored) {
            }

            Iterator<MessageQueue> iterator = processQueueTable.keySet().iterator();
            while (iterator.hasNext()) {
                MessageQueue mq = iterator.next();
                Long offset = offsetTable.get(mq);
                if (topic.equals(mq.getTopic()) && offset != null) {
                    try {
                        consumer.updateConsumeOffset(mq, offset);
                        consumer.getRebalanceImpl().removeUnnecessaryMessageQueue(mq, processQueueTable.get(mq));
                        iterator.remove();
                    } catch (Exception e) {
                        log.warn("reset offset failed. group={}, {}", group, mq, e);
                    }
                }
            }
        } finally {
            if (consumer != null) {
                consumer.resume();
            }
        }
    }
@Test
    public void testResetOffset() throws IllegalAccessException {
        topicRouteTable.put(topic, createTopicRouteData());
        brokerAddrTable.put(defaultBroker, createBrokerAddrMap());
        consumerTable.put(group, createMQConsumerInner());
        Map<MessageQueue, Long> offsetTable = new HashMap<>();
        offsetTable.put(createMessageQueue(), 0L);
        mqClientInstance.resetOffset(topic, group, offsetTable);
        Field consumerTableField = FieldUtils.getDeclaredField(mqClientInstance.getClass(), "consumerTable", true);
        ConcurrentMap<String, MQConsumerInner> consumerTable = (ConcurrentMap<String, MQConsumerInner>) consumerTableField.get(mqClientInstance);
        DefaultMQPushConsumerImpl consumer = (DefaultMQPushConsumerImpl) consumerTable.get(group);
        verify(consumer).suspend();
        verify(consumer).resume();
        verify(consumer, times(1))
                .updateConsumeOffset(
                        any(MessageQueue.class),
                        eq(0L));
    }