@Test public void eventuallyStopRetrying() { long lowerBound = 10; long upperBound = 25; insertValuesWithinBounds(lowerBound, upperBound, source); PaxosStateLog<PaxosValue> targetMock = mock(PaxosStateLog.class); Exception expectedException = new RuntimeException("We failed"); doThrow(expectedException).when(targetMock).writeBatchOfRounds(any()); ImmutableMigrationContext<PaxosValue> context = ImmutableMigrationContext.<PaxosValue>builder() .sourceLog(source) .destinationLog(targetMock) .hydrator(PaxosValue.BYTES_HYDRATOR) .migrationState(migrationState) .migrateFrom(lowerBound) .build(); assertThatThrownBy(() -> PaxosStateLogMigrator.migrateAndReturnCutoff(context)) .isEqualTo(expectedException); }
public static <V extends Persistable & Versionable> long migrateAndReturnCutoff(MigrationContext<V> context) { PaxosStateLogMigrator<V> migrator = new PaxosStateLogMigrator<>(context.sourceLog(), context.destinationLog()); if (!context.migrationState().isInMigratedState()) { long cutoff = calculateCutoff(context); migrator.runMigration(cutoff, context.hydrator()); context.migrationState().setCutoff(cutoff); context.migrationState().migrateToMigratedState(); return cutoff; } return context.migrationState().getCutoff(); }
PaxosStateLogMigrator { public static <V extends Persistable & Versionable> long migrateAndReturnCutoff(MigrationContext<V> context) { PaxosStateLogMigrator<V> migrator = new PaxosStateLogMigrator<>(context.sourceLog(), context.destinationLog()); if (!context.migrationState().isInMigratedState()) { long cutoff = calculateCutoff(context); migrator.runMigration(cutoff, context.hydrator()); context.migrationState().setCutoff(cutoff); context.migrationState().migrateToMigratedState(); return cutoff; } return context.migrationState().getCutoff(); } }
PaxosStateLogMigrator { public static <V extends Persistable & Versionable> long migrateAndReturnCutoff(MigrationContext<V> context) { PaxosStateLogMigrator<V> migrator = new PaxosStateLogMigrator<>(context.sourceLog(), context.destinationLog()); if (!context.migrationState().isInMigratedState()) { long cutoff = calculateCutoff(context); migrator.runMigration(cutoff, context.hydrator()); context.migrationState().setCutoff(cutoff); context.migrationState().migrateToMigratedState(); return cutoff; } return context.migrationState().getCutoff(); } private  PaxosStateLogMigrator(PaxosStateLog<V> sourceLog, PaxosStateLog<V> destinationLog); }
PaxosStateLogMigrator { public static <V extends Persistable & Versionable> long migrateAndReturnCutoff(MigrationContext<V> context) { PaxosStateLogMigrator<V> migrator = new PaxosStateLogMigrator<>(context.sourceLog(), context.destinationLog()); if (!context.migrationState().isInMigratedState()) { long cutoff = calculateCutoff(context); migrator.runMigration(cutoff, context.hydrator()); context.migrationState().setCutoff(cutoff); context.migrationState().migrateToMigratedState(); return cutoff; } return context.migrationState().getCutoff(); } private  PaxosStateLogMigrator(PaxosStateLog<V> sourceLog, PaxosStateLog<V> destinationLog); static long migrateAndReturnCutoff(MigrationContext<V> context); }
PaxosStateLogMigrator { public static <V extends Persistable & Versionable> long migrateAndReturnCutoff(MigrationContext<V> context) { PaxosStateLogMigrator<V> migrator = new PaxosStateLogMigrator<>(context.sourceLog(), context.destinationLog()); if (!context.migrationState().isInMigratedState()) { long cutoff = calculateCutoff(context); migrator.runMigration(cutoff, context.hydrator()); context.migrationState().setCutoff(cutoff); context.migrationState().migrateToMigratedState(); return cutoff; } return context.migrationState().getCutoff(); } private  PaxosStateLogMigrator(PaxosStateLog<V> sourceLog, PaxosStateLog<V> destinationLog); static long migrateAndReturnCutoff(MigrationContext<V> context); static final int SAFETY_BUFFER; }