@Test void migrationShouldBeIdempotentWhenRunMultipleTimes() throws ExecutionException, InterruptedException { IntStream.range(0, MAPPING_COUNT) .forEach(i -> cassandraRecipientRewriteTableDAO .addMapping(MappingSource.parse("source" + i + "@domain"), MAPPING).block()); ConcurrentTestRunner.builder() .operation((threadNumber, step) -> migration.apply()) .threadCount(THREAD_COUNT) .operationCount(OPERATION_COUNT) .runSuccessfullyWithin(Duration.ofMinutes(1)); assertThat(cassandraMappingsSourcesDAO.retrieveSources(MAPPING).collectList().block()) .hasSize(MAPPING_COUNT); }
@Override public void apply() { cassandraRecipientRewriteTableDAO.getAllMappings() .flatMap(this::migrate) .then(Mono.fromRunnable(() -> { if (errorMappingsCount.get() > 0) { throw new MigrationException("MappingsSourcesMigration failed"); } })) .doOnError(t -> LOGGER.error("Error while migrating mappings sources", t)) .block(); }
MappingsSourcesMigration implements Migration { @Override public void apply() { cassandraRecipientRewriteTableDAO.getAllMappings() .flatMap(this::migrate) .then(Mono.fromRunnable(() -> { if (errorMappingsCount.get() > 0) { throw new MigrationException("MappingsSourcesMigration failed"); } })) .doOnError(t -> LOGGER.error("Error while migrating mappings sources", t)) .block(); } }
MappingsSourcesMigration implements Migration { @Override public void apply() { cassandraRecipientRewriteTableDAO.getAllMappings() .flatMap(this::migrate) .then(Mono.fromRunnable(() -> { if (errorMappingsCount.get() > 0) { throw new MigrationException("MappingsSourcesMigration failed"); } })) .doOnError(t -> LOGGER.error("Error while migrating mappings sources", t)) .block(); } @Inject  MappingsSourcesMigration(CassandraRecipientRewriteTableDAO cassandraRecipientRewriteTableDAO,
                                    CassandraMappingsSourcesDAO cassandraMappingsSourcesDAO); }
MappingsSourcesMigration implements Migration { @Override public void apply() { cassandraRecipientRewriteTableDAO.getAllMappings() .flatMap(this::migrate) .then(Mono.fromRunnable(() -> { if (errorMappingsCount.get() > 0) { throw new MigrationException("MappingsSourcesMigration failed"); } })) .doOnError(t -> LOGGER.error("Error while migrating mappings sources", t)) .block(); } @Inject  MappingsSourcesMigration(CassandraRecipientRewriteTableDAO cassandraRecipientRewriteTableDAO,
                                    CassandraMappingsSourcesDAO cassandraMappingsSourcesDAO); @Override void apply(); @Override Task asTask(); }
MappingsSourcesMigration implements Migration { @Override public void apply() { cassandraRecipientRewriteTableDAO.getAllMappings() .flatMap(this::migrate) .then(Mono.fromRunnable(() -> { if (errorMappingsCount.get() > 0) { throw new MigrationException("MappingsSourcesMigration failed"); } })) .doOnError(t -> LOGGER.error("Error while migrating mappings sources", t)) .block(); } @Inject  MappingsSourcesMigration(CassandraRecipientRewriteTableDAO cassandraRecipientRewriteTableDAO,
                                    CassandraMappingsSourcesDAO cassandraMappingsSourcesDAO); @Override void apply(); @Override Task asTask(); static final TaskType TYPE; }