ConsumerOffsetManager extends ConfigManager { public void commitOffset(final String group, final String topic, final int queueId, final long offset) { String key = topic + TOPIC_GROUP_SEPARATOR + group; this.commitOffset(key, queueId, offset); } ConsumerOffsetManager();  ConsumerOffsetManager(BrokerController brokerController); void scanUnsubscribedTopic(); Set<String> whichTopicByConsumer(final String group); Set<String> whichGroupByTopic(final String topic); void commitOffset(final String group, final String topic, final int queueId, final long offset); long queryOffset(final String group, final String topic, final int queueId); String encode(); String encode(final boolean prettyFormat); @Override void decode(String jsonString); @Override String configFilePath(); ConcurrentHashMap<String, ConcurrentHashMap<Integer, Long>> getOffsetTable(); void setOffsetTable(ConcurrentHashMap<String, ConcurrentHashMap<Integer, Long>> offsetTable); Map<Integer, Long> queryMinOffsetInAllGroup(final String topic, final String filterGroups); Map<Integer, Long> queryOffset(final String group, final String topic); void cloneOffset(final String srcGroup, final String destGroup, final String topic);  }
@Test public void test_flushConsumerOffset() throws Exception { BrokerController brokerController = new BrokerController( new BrokerConfig(), new NettyServerConfig(), new NettyClientConfig(), new MessageStoreConfig()); boolean initResult = brokerController.initialize(); System.out.println("initialize " + initResult); brokerController.start(); ConsumerOffsetManager consumerOffsetManager = new ConsumerOffsetManager(brokerController); Random random = new Random(); for (int i = 0; i < 100; i++) { String group = "DIANPU_GROUP_" + i; for (int id = 0; id < 16; id++) { consumerOffsetManager.commitOffset(group, "TOPIC_A", id, random.nextLong() % 1024 * 1024 * 1024); consumerOffsetManager.commitOffset(group, "TOPIC_B", id, random.nextLong() % 1024 * 1024 * 1024); consumerOffsetManager.commitOffset(group, "TOPIC_C", id, random.nextLong() % 1024 * 1024 * 1024); } } consumerOffsetManager.persist(); brokerController.shutdown(); }