### Question:
ConsumerCommitServiceImpl implements ConsumerCommitService, BrokerTimerService { protected void commitOffset() { log.info("doSubmitOffset"); while (isRunning) { doCommit(); Util.sleep(soaConfig.getCommitSleepTime()); } }  @Override void startBroker(); @Override CommitOffsetResponse commitOffset(CommitOffsetRequest request); @Override void stopBroker(); @Override String info(); @Override Map<Long, ConsumerQueueVersionDto> getCache();  }### Answer:
@Test public void commitOffsetTest() { CommitOffsetRequest request = new CommitOffsetRequest(); request.setFlag(1); List<ConsumerQueueVersionDto> queueOffsets = buildQueueOffsets(); request.setQueueOffsets(queueOffsets); when(queueOffsetService.commitOffset(any(QueueOffsetEntity.class))).thenReturn(1); consumerCommitServiceImpl.commitOffset(request); verify(queueOffsetService).commitOffset(any(QueueOffsetEntity.class)); queueOffsets = buildQueueOffsets(); request.setQueueOffsets(queueOffsets); request.setFlag(0); queueOffsets.get(0).setOffsetVersion(queueOffsets.get(0).getOffsetVersion() + 1); consumerCommitServiceImpl.commitOffset(request); verify(queueOffsetService,times(1)).commitOffset(any(QueueOffsetEntity.class)); queueOffsets.get(0).setOffset(queueOffsets.get(0).getOffset() + 1); consumerCommitServiceImpl.commitOffset(request); verify(queueOffsetService,times(1)).commitOffset(any(QueueOffsetEntity.class)); }