@Test public void testPartialApplyForBadOps() throws Exception { CuratorOp op1 = client.transactionOp().create().forPath("/test", "something".getBytes()); CuratorOp op2 = client.transactionOp().create().forPath("/a/b/c"); Migration m1 = () -> Collections.singletonList(op1); Migration m2 = () -> Collections.singletonList(op2); MigrationSet migrationSet = MigrationSet.build("1", Arrays.asList(m1, m2)); try { complete(manager.migrate(migrationSet)); Assert.fail("Should throw"); } catch ( Throwable e ) { Assert.assertTrue(Throwables.getRootCause(e) instanceof KeeperException.NoNodeException); } Assert.assertNull(client.unwrap().checkExists().forPath("/test")); }
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);  }