### Question:
ReconTaskControllerImpl implements ReconTaskController { @Override public synchronized void consumeOMEvents(OMUpdateEventBatch events, OMMetadataManager omMetadataManager) throws InterruptedException { try { if (!events.isEmpty()) { Collection<Callable<Pair<String, Boolean>>> tasks = new ArrayList<>(); for (Map.Entry<String, ReconOmTask> taskEntry : reconOmTasks.entrySet()) { ReconOmTask task = taskEntry.getValue(); Collection<String> tables = task.getTaskTables(); tasks.add(() -> task.process(events.filter(tables))); } List<Future<Pair<String, Boolean>>> results = executorService.invokeAll(tasks); List<String> failedTasks = processTaskResults(results, events); List<String> retryFailedTasks = new ArrayList<>(); if (!failedTasks.isEmpty()) { tasks.clear(); for (String taskName : failedTasks) { ReconOmTask task = reconOmTasks.get(taskName); Collection<String> tables = task.getTaskTables(); tasks.add(() -> task.process(events.filter(tables))); } results = executorService.invokeAll(tasks); retryFailedTasks = processTaskResults(results, events); } if (!retryFailedTasks.isEmpty()) { tasks.clear(); for (String taskName : failedTasks) { ReconOmTask task = reconOmTasks.get(taskName); tasks.add(() -> task.reprocess(omMetadataManager)); } results = executorService.invokeAll(tasks); List<String> reprocessFailedTasks = processTaskResults(results, events); ignoreFailedTasks(reprocessFailedTasks); } } } catch (ExecutionException e) { LOG.error("Unexpected error : ", e); } } @Inject  ReconTaskControllerImpl(OzoneConfiguration configuration,
                                 ReconTaskStatusDao reconTaskStatusDao,
                                 Set<ReconOmTask> tasks); @Override void registerTask(ReconOmTask task); @Override synchronized void consumeOMEvents(OMUpdateEventBatch events,
                              OMMetadataManager omMetadataManager); @Override synchronized void reInitializeTasks(
      ReconOMMetadataManager omMetadataManager); @Override Map<String, ReconOmTask> getRegisteredTasks(); @Override ReconTaskStatusDao getReconTaskStatusDao(); @Override synchronized void start(); @Override synchronized void stop();  }### Answer:
@Test public void testConsumeOMEvents() throws Exception { ReconOmTask reconOmTaskMock = getMockTask("MockTask"); when(reconOmTaskMock.process(any(OMUpdateEventBatch.class))) .thenReturn(new ImmutablePair<>("MockTask", true)); reconTaskController.registerTask(reconOmTaskMock); OMUpdateEventBatch omUpdateEventBatchMock = mock(OMUpdateEventBatch.class); when(omUpdateEventBatchMock.getLastSequenceNumber()).thenReturn(100L); when(omUpdateEventBatchMock.isEmpty()).thenReturn(false); when(omUpdateEventBatchMock.filter(Collections.singleton("MockTable"))) .thenReturn(omUpdateEventBatchMock); long startTime = System.currentTimeMillis(); reconTaskController.consumeOMEvents( omUpdateEventBatchMock, mock(OMMetadataManager.class)); verify(reconOmTaskMock, times(1)) .process(any()); long endTime = System.currentTimeMillis(); reconTaskStatusDao = getDao(ReconTaskStatusDao.class); ReconTaskStatus reconTaskStatus = reconTaskStatusDao.findById("MockTask"); long taskTimeStamp = reconTaskStatus.getLastUpdatedTimestamp(); long seqNumber = reconTaskStatus.getLastUpdatedSeqNumber(); Assert.assertTrue(startTime <= taskTimeStamp && taskTimeStamp <= endTime); Assert.assertEquals(seqNumber, omUpdateEventBatchMock.getLastSequenceNumber()); }