@Override
    public void persistAll(Set<MessageQueue> mqs) {
        if (null == mqs || mqs.isEmpty()) {
            return;
        }
        OffsetSerializeWrapper offsetSerializeWrapper = null;
        try {
            offsetSerializeWrapper = readLocalOffset();
        } catch (MQClientException e) {
            log.error("readLocalOffset exception", e);
            return;
        }

        if (offsetSerializeWrapper == null) {
            offsetSerializeWrapper = new OffsetSerializeWrapper();
        }
        for (Map.Entry<MessageQueue, ControllableOffset> entry : this.offsetTable.entrySet()) {
            if (mqs.contains(entry.getKey())) {
                AtomicLong offset = new AtomicLong(entry.getValue().getOffset());
                offsetSerializeWrapper.getOffsetTable().put(entry.getKey(), offset);
            }
        }

        String jsonString = offsetSerializeWrapper.toJson(true);
        if (jsonString != null) {
            try {
                MixAll.string2File(jsonString, this.storePath);
            } catch (IOException e) {
                log.error("persistAll consumer offset Exception, " + this.storePath, e);
            }
        }
    }
@Test
    public void testPersistAll() throws Exception {
        OffsetStore offsetStore = new LocalFileOffsetStore(mQClientFactory, group);

        MessageQueue messageQueue0 = new MessageQueue(topic, brokerName, 0);
        offsetStore.updateOffset(messageQueue0, 1024, false);
        offsetStore.persistAll(new HashSet<MessageQueue>(Collections.singletonList(messageQueue0)));
        assertThat(offsetStore.readOffset(messageQueue0, ReadOffsetType.READ_FROM_STORE)).isEqualTo(1024);

        MessageQueue messageQueue1 = new MessageQueue(topic, brokerName, 1);
        MessageQueue messageQueue2 = new MessageQueue(topic, brokerName, 2);
        offsetStore.updateOffset(messageQueue1, 1025, false);
        offsetStore.updateOffset(messageQueue2, 1026, false);
        offsetStore.persistAll(new HashSet<MessageQueue>(Arrays.asList(messageQueue1, messageQueue2)));

        assertThat(offsetStore.readOffset(messageQueue0, ReadOffsetType.READ_FROM_STORE)).isEqualTo(1024);
        assertThat(offsetStore.readOffset(messageQueue1, ReadOffsetType.READ_FROM_STORE)).isEqualTo(1025);
        assertThat(offsetStore.readOffset(messageQueue2, ReadOffsetType.READ_FROM_STORE)).isEqualTo(1026);
    }