class ConsumerCommitServiceImpl implements ConsumerCommitService, BrokerTimerService {

	protected void doCommit() {
		Map<Long, OffsetVersionEntity> offsetVersionMap = queueOffsetService.getOffsetVersion();
		Map<Long, ConsumerQueueVersionDto> map = new HashMap<>(mapAppPolling.get());
		if (map.size() > 0) {
			final int size = map.size();
			Transaction transaction = Tracer.newTransaction("Timer-service", "commit");
			try {
				int countSize = map.size() < commitThreadSize ? map.size() : commitThreadSize;
				if (countSize == 1) {
					for (Map.Entry<Long, ConsumerQueueVersionDto> entry : map.entrySet()) {
						doCommitOffset(entry.getValue(), 0, offsetVersionMap, size);
					}
				} else {
					CountDownLatch countDownLatch = new CountDownLatch(countSize);
					for (Map.Entry<Long, ConsumerQueueVersionDto> entry : map.entrySet()) {
						executorRun.execute(new Runnable() {
							@Override
							public void run() {
								doCommitOffset(entry.getValue(), 0, offsetVersionMap, size);
								countDownLatch.countDown();
							}
						});
					}
					countDownLatch.await();
				}
				transaction.setStatus(Transaction.SUCCESS);
			} catch (Exception e) {
				transaction.setStatus(e);
			} finally {
				transaction.complete();
			}
		}
	}

}