@Test public void testStaged() { Migration m1 = () -> Arrays.asList(v1opA, v1opB); MigrationSet migrationSet = MigrationSet.build("1", Collections.singletonList(m1)); complete(manager.migrate(migrationSet)); ModeledFramework<ModelV1> v1Client = ModeledFramework.wrap(client, v1Spec); complete(v1Client.read(), (m, e) -> Assert.assertEquals(m.getName(), "Test")); Migration m2 = () -> Collections.singletonList(v2op); migrationSet = MigrationSet.build("1", Arrays.asList(m1, m2)); complete(manager.migrate(migrationSet)); ModeledFramework<ModelV2> v2Client = ModeledFramework.wrap(client, v2Spec); complete(v2Client.read(), (m, e) -> { Assert.assertEquals(m.getName(), "Test 2"); Assert.assertEquals(m.getAge(), 10); }); Migration m3 = () -> Collections.singletonList(v3op); migrationSet = MigrationSet.build("1", Arrays.asList(m1, m2, m3)); complete(manager.migrate(migrationSet)); ModeledFramework<ModelV3> v3Client = ModeledFramework.wrap(client, v3Spec); complete(v3Client.read(), (m, e) -> { Assert.assertEquals(m.getAge(), 30); Assert.assertEquals(m.getFirstName(), "One"); Assert.assertEquals(m.getLastName(), "Two"); }); }
public CompletionStage<Void> migrate(MigrationSet set) { InterProcessLock lock = new InterProcessSemaphoreMutex(client.unwrap(), ZKPaths.makePath(lockPath, set.id())); CompletionStage<Void> lockStage = lockAsync(lock, lockMax.toMillis(), TimeUnit.MILLISECONDS, executor); return lockStage.thenCompose(__ -> runMigrationInLock(lock, set)); }
MigrationManager { public CompletionStage<Void> migrate(MigrationSet set) { InterProcessLock lock = new InterProcessSemaphoreMutex(client.unwrap(), ZKPaths.makePath(lockPath, set.id())); CompletionStage<Void> lockStage = lockAsync(lock, lockMax.toMillis(), TimeUnit.MILLISECONDS, executor); return lockStage.thenCompose(__ -> runMigrationInLock(lock, set)); } }
MigrationManager { public CompletionStage<Void> migrate(MigrationSet set) { InterProcessLock lock = new InterProcessSemaphoreMutex(client.unwrap(), ZKPaths.makePath(lockPath, set.id())); CompletionStage<Void> lockStage = lockAsync(lock, lockMax.toMillis(), TimeUnit.MILLISECONDS, executor); return lockStage.thenCompose(__ -> runMigrationInLock(lock, set)); } MigrationManager(AsyncCuratorFramework client, String lockPath, String metaDataPath, Executor executor, Duration lockMax); }
MigrationManager { public CompletionStage<Void> migrate(MigrationSet set) { InterProcessLock lock = new InterProcessSemaphoreMutex(client.unwrap(), ZKPaths.makePath(lockPath, set.id())); CompletionStage<Void> lockStage = lockAsync(lock, lockMax.toMillis(), TimeUnit.MILLISECONDS, executor); return lockStage.thenCompose(__ -> runMigrationInLock(lock, set)); } MigrationManager(AsyncCuratorFramework client, String lockPath, String metaDataPath, Executor executor, Duration lockMax); CompletionStage<Void> migrate(MigrationSet set); }
MigrationManager { public CompletionStage<Void> migrate(MigrationSet set) { InterProcessLock lock = new InterProcessSemaphoreMutex(client.unwrap(), ZKPaths.makePath(lockPath, set.id())); CompletionStage<Void> lockStage = lockAsync(lock, lockMax.toMillis(), TimeUnit.MILLISECONDS, executor); return lockStage.thenCompose(__ -> runMigrationInLock(lock, set)); } MigrationManager(AsyncCuratorFramework client, String lockPath, String metaDataPath, Executor executor, Duration lockMax); CompletionStage<Void> migrate(MigrationSet set);  }