elasticsearch-agent/src/main/java/com/epam/pipeline/elasticsearchagent/service/ElasticsearchAgentService.java
@Scheduled(fixedDelayString = "${sync.scheduler.delay}")
    @Transactional(propagation = Propagation.REQUIRED)
    public void startElasticsearchAgent() {
        log.debug("Start scheduled database changes...");

        LocalDateTime lastSyncTime = getLastSyncTime();
        LocalDateTime syncStart = LocalDateTime.now(Clock.systemUTC());

        List<CompletableFuture<Void>> results = synchronizers.stream()
                .map(synchronizer -> CompletableFuture.runAsync(() -> {
                    LocalDateTime startSyncTime = LocalDateTime.now(Clock.systemUTC());
                    log.debug("Synchronizer {} starts work at {} ", synchronizer.getClass().getSimpleName(),
                            startSyncTime);
                    synchronizer.synchronize(lastSyncTime, syncStart);
                    log.debug("Synchronizer {} stops work at {}. Duration is {} seconds.",
                            synchronizer.getClass().getSimpleName(),
                            LocalDateTime.now(Clock.systemUTC()),
                            Duration.between(startSyncTime, LocalDateTime.now(Clock.systemUTC()))
                                    .abs()
                                    .getSeconds());
                }, elasticsearchAgentThreadPool)
                .exceptionally(throwable -> {
                    log.warn("Exception while trying to send data to Elasticsearch service", throwable);
                    log.debug("Synchronizer {} stops work at {}.",
                            synchronizer.getClass().getSimpleName(), LocalDateTime.now(Clock.systemUTC()));
                    return null;
                })).collect(Collectors.toList());
        try {
            CompletableFuture.allOf(results.toArray(new CompletableFuture[0])).get();
            Files.write(Paths.get(lastSynchronizationTimeFilePath),
                    (syncStart.toString() + System.lineSeparator()).getBytes(),
                    StandardOpenOption.APPEND, StandardOpenOption.CREATE);
            log.debug("Finished synchronization...");
        } catch (IOException | InterruptedException | ExecutionException e) {
            log.error("An error occurred synchronization: {}", e.getMessage());
            log.error(e.getMessage(), e);
        }

        log.debug("Stop Elasticsearch agent.");
    }
/**
     * Scheduled task to monitor changes in the database,
     * to retrieve and send data to Elasticsearch service
     * to indexes the data as documents inside Elasticsearch
     */
Scheduled task to monitor changes in the database,
to retrieve and send data to Elasticsearch service
to indexes the data as documents inside Elasticsearch
Scheduled task to monitor changes in the database to retrieve and send data to Elasticsearch service to indexes the data as documents inside Elasticsearch
@Scheduled(fixedDelayString = "${sync.scheduler.delay}")
    @Transactional(propagation = Propagation.REQUIRED)
    public void startElasticsearchAgent() {
        log.debug("Start scheduled database changes...");
        LocalDateTime lastSyncTime = getLastSyncTime();
        LocalDateTime syncStart = LocalDateTime.now(Clock.systemUTC());
        List<CompletableFuture<Void>> results = synchronizers.stream()
                .map(synchronizer -> CompletableFuture.runAsync(() -> {
                    LocalDateTime startSyncTime = LocalDateTime.now(Clock.systemUTC());
                    log.debug("Synchronizer {} starts work at {} ", synchronizer.getClass().getSimpleName(),
                            startSyncTime);
                    synchronizer.synchronize(lastSyncTime, syncStart);
                    log.debug("Synchronizer {} stops work at {}. Duration is {} seconds.",
                            synchronizer.getClass().getSimpleName(),
                            LocalDateTime.now(Clock.systemUTC()),
                            Duration.between(startSyncTime, LocalDateTime.now(Clock.systemUTC()))
                                    .abs()
                                    .getSeconds());
                }, elasticsearchAgentThreadPool)
                .exceptionally(throwable -> {
                    log.warn("Exception while trying to send data to Elasticsearch service", throwable);
                    log.debug("Synchronizer {} stops work at {}.",
                            synchronizer.getClass().getSimpleName(), LocalDateTime.now(Clock.systemUTC()));
                    return null;
                })).collect(Collectors.toList());
        try {
            CompletableFuture.allOf(results.toArray(new CompletableFuture[0])).get();
            Files.write(Paths.get(lastSynchronizationTimeFilePath),
                    (syncStart.toString() + System.lineSeparator()).getBytes(),
                    StandardOpenOption.APPEND, StandardOpenOption.CREATE);
            log.debug("Finished synchronization...");
        } catch (IOException | InterruptedException | ExecutionException e) {
            log.error("An error occurred synchronization: {}", e.getMessage());
            log.error(e.getMessage(), e);
        }
        log.debug("Stop Elasticsearch agent.");
    }
@ Scheduled ( fixedDelayString = "${sync.scheduler.delay}" ) @ Transactional ( propagation = Propagation . REQUIRED ) public void startElasticsearchAgent ( ) { log . debug ( "Start scheduled database changes..." ) ; LocalDateTime lastSyncTime = getLastSyncTime ( ) ; LocalDateTime syncStart = LocalDateTime . now ( Clock . systemUTC ( ) ) ; List < CompletableFuture < Void > > results = synchronizers . stream ( ) . map ( synchronizer -> CompletableFuture . runAsync ( ( ) -> { LocalDateTime startSyncTime = LocalDateTime . now ( Clock . systemUTC ( ) ) ; log . debug ( "Synchronizer {} starts work at {} " , synchronizer . getClass ( ) . getSimpleName ( ) , startSyncTime ) ; synchronizer . synchronize ( lastSyncTime , syncStart ) ; log . debug ( "Synchronizer {} stops work at {}. Duration is {} seconds." , synchronizer . getClass ( ) . getSimpleName ( ) , LocalDateTime . now ( Clock . systemUTC ( ) ) , Duration . between ( startSyncTime , LocalDateTime . now ( Clock . systemUTC ( ) ) ) . abs ( ) . getSeconds ( ) ) ; } , elasticsearchAgentThreadPool ) . exceptionally ( throwable -> { log . warn ( "Exception while trying to send data to Elasticsearch service" , throwable ) ; log . debug ( "Synchronizer {} stops work at {}." , synchronizer . getClass ( ) . getSimpleName ( ) , LocalDateTime . now ( Clock . systemUTC ( ) ) ) ; return null ; } ) ) . collect ( Collectors . toList ( ) ) ; try { CompletableFuture . allOf ( results . toArray ( new CompletableFuture [ 0 ] ) ) . get ( ) ; Files . write ( Paths . get ( lastSynchronizationTimeFilePath ) , ( syncStart . toString ( ) + System . lineSeparator ( ) ) . getBytes ( ) , StandardOpenOption . APPEND , StandardOpenOption . CREATE ) ; log . debug ( "Finished synchronization..." ) ; } catch ( IOException | InterruptedException | ExecutionException e ) { log . error ( "An error occurred synchronization: {}" , e . getMessage ( ) ) ; log . error ( e . getMessage ( ) , e ) ; } log . debug ( "Stop Elasticsearch agent." ) ; }
Scheduled task to monitor changes in the database,
 to retrieve and send data to Elasticsearch service
 to indexes the data as documents inside Elasticsearch
Scheduled task to monitor changes in the database to retrieve and send data to Elasticsearch service to indexes the data as documents inside Elasticsearch