public void resetOffsetByTimeStamp(long timeStamp) throws MQClientException {
        for (String topic : rebalanceImpl.getSubscriptionInner().keySet()) {
            Set<MessageQueue> mqs = rebalanceImpl.getTopicSubscribeInfoTable().get(topic);
            if (CollectionUtils.isNotEmpty(mqs)) {
                Map<MessageQueue, Long> offsetTable = new HashMap<>(mqs.size(), 1);
                for (MessageQueue mq : mqs) {
                    long offset = searchOffset(mq, timeStamp);
                    offsetTable.put(mq, offset);
                }
                this.mQClientFactory.resetOffset(topic, groupName(), offsetTable);
            }
        }
    }
@Test
    public void testResetOffsetByTimeStamp() throws MQClientException {
        ConcurrentMap<String, SubscriptionData> subscriptionDataMap = new ConcurrentHashMap<>();
        subscriptionDataMap.put(defaultTopic, new SubscriptionData());
        when(rebalanceImpl.getSubscriptionInner()).thenReturn(subscriptionDataMap);
        defaultMQPushConsumerImpl.resetOffsetByTimeStamp(System.currentTimeMillis());
        verify(mQClientFactory).resetOffset(eq(defaultTopic), any(), any());
    }