### Question:
PipelineSynchronizer implements ElasticsearchSynchronizer { @Override public void synchronize(final LocalDateTime lastSyncTime, final LocalDateTime syncStart) { log.debug("Started pipeline with code synchronization"); final List<PipelineEvent> pipelineEvents = EventProcessorUtils.mergeEvents( pipelineEventDao.loadPipelineEventsByObjectType(PipelineEvent.ObjectType.PIPELINE, syncStart)); final List<PipelineEvent> pipelineCodeEvents = pipelineEventDao .loadPipelineEventsByObjectType(PipelineEvent.ObjectType.PIPELINE_CODE, syncStart); if (CollectionUtils.isEmpty(pipelineEvents) && CollectionUtils.isEmpty(pipelineCodeEvents)) { log.debug("{} and {} entities for synchronization were not found.", PipelineEvent.ObjectType.PIPELINE, PipelineEvent.ObjectType.PIPELINE_CODE); return; } Stream.of(pipelineEvents, pipelineCodeEvents) .flatMap(Collection::stream) .collect(Collectors.groupingBy(PipelineEvent::getObjectId)) .forEach((id, events) -> synchronizePipelineEvents(id, events, syncStart)); log.debug("Successfully finished {} and {} synchronization.", PipelineEvent.ObjectType.PIPELINE, PipelineEvent.ObjectType.PIPELINE_CODE); } PipelineSynchronizer(
            final PipelineEventDao pipelineEventDao,
            final @Value("${sync.pipeline.index.mapping}") String pipelineIndexMappingFile,
            final @Value("${sync.pipeline-code.index.mapping}") String pipelineCodeIndexMappingFile,
            final @Value("${sync.index.common.prefix}") String indexPrefix,
            final @Value("${sync.pipeline.index.name}") String pipelineIndexName,
            final @Value("${sync.pipeline-code.index.name}") String pipelineCodeIndexName,
            final @Value("${sync.pipeline-code.index.paths}") String pipelineFileIndexPaths,
            final @Value("${sync.pipeline-code.bulk.insert.size}") Integer bulkInsertSize,
            final CloudPipelineAPIClient cloudPipelineAPIClient,
            final ElasticsearchServiceClient elasticsearchServiceClient,
            final ElasticIndexService indexService,
            final PipelineLoader loader,
            final PipelineMapper mapper,
            final PipelineCodeHandler codeHandler,
            final BulkResponsePostProcessor bulkResponsePostProcessor); @Override void synchronize(final LocalDateTime lastSyncTime, final LocalDateTime syncStart);  }### Answer:
@Test void shouldStartSynchronizeTest() throws InterruptedException { TimeUnit.SECONDS.sleep(2); verify(pipelineSynchronizer, atLeast(1)).synchronize(any(), any()); }