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);  }
@Test public void testBasic() { Migration m1 = () -> Arrays.asList(v1opA, v1opB); Migration m2 = () -> Collections.singletonList(v2op); Migration m3 = () -> Collections.singletonList(v3op); MigrationSet 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"); }); int count = manager.debugCount.get(); complete(manager.migrate(migrationSet)); Assert.assertEquals(manager.debugCount.get(), count); }
@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"); }); }
@Test public void testDocExample() throws Exception { CuratorOp op1 = client.transactionOp().create().forPath("/parent"); CuratorOp op2 = client.transactionOp().create().forPath("/parent/one"); CuratorOp op3 = client.transactionOp().create().forPath("/parent/two"); CuratorOp op4 = client.transactionOp().create().forPath("/parent/three"); CuratorOp op5 = client.transactionOp().create().forPath("/main", "hey".getBytes()); Migration initialMigration = () -> Arrays.asList(op1, op2, op3, op4, op5); MigrationSet migrationSet = MigrationSet.build("main", Collections.singletonList(initialMigration)); complete(manager.migrate(migrationSet)); Assert.assertNotNull(client.unwrap().checkExists().forPath("/parent/three")); Assert.assertEquals(client.unwrap().getData().forPath("/main"), "hey".getBytes()); CuratorOp newOp1 = client.transactionOp().create().forPath("/new"); CuratorOp newOp2 = client.transactionOp().delete().forPath("/main"); Migration newMigration = () -> Arrays.asList(newOp1, newOp2); migrationSet = MigrationSet.build("main", Arrays.asList(initialMigration, newMigration)); complete(manager.migrate(migrationSet)); Assert.assertNull(client.unwrap().checkExists().forPath("/main")); }
@Test public void testChecksumDataError() { CuratorOp op1 = client.transactionOp().create().forPath("/test"); CuratorOp op2 = client.transactionOp().create().forPath("/test/bar", "first".getBytes()); Migration migration = () -> Arrays.asList(op1, op2); MigrationSet migrationSet = MigrationSet.build("1", Collections.singletonList(migration)); complete(manager.migrate(migrationSet)); CuratorOp op2Changed = client.transactionOp().create().forPath("/test/bar", "second".getBytes()); migration = () -> Arrays.asList(op1, op2Changed); migrationSet = MigrationSet.build("1", Collections.singletonList(migration)); try { complete(manager.migrate(migrationSet)); Assert.fail("Should throw"); } catch ( Throwable e ) { Assert.assertTrue(Throwables.getRootCause(e) instanceof MigrationException); } }
@Test public void testChecksumPathError() { CuratorOp op1 = client.transactionOp().create().forPath("/test2"); CuratorOp op2 = client.transactionOp().create().forPath("/test2/bar"); Migration migration = () -> Arrays.asList(op1, op2); MigrationSet migrationSet = MigrationSet.build("1", Collections.singletonList(migration)); complete(manager.migrate(migrationSet)); CuratorOp op2Changed = client.transactionOp().create().forPath("/test/bar"); migration = () -> Arrays.asList(op1, op2Changed); migrationSet = MigrationSet.build("1", Collections.singletonList(migration)); try { complete(manager.migrate(migrationSet)); Assert.fail("Should throw"); } catch ( Throwable e ) { Assert.assertTrue(Throwables.getRootCause(e) instanceof MigrationException); } }
@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")); }
@Test public void testTransactionForBadOps() throws Exception { CuratorOp op1 = client.transactionOp().create().forPath("/test2", "something".getBytes()); CuratorOp op2 = client.transactionOp().create().forPath("/a/b/c/d"); Migration migration = () -> Arrays.asList(op1, op2); MigrationSet migrationSet = MigrationSet.build("1", Collections.singletonList(migration)); 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")); }
@Test public void testConcurrency1() throws Exception { CuratorOp op1 = client.transactionOp().create().forPath("/test"); CuratorOp op2 = client.transactionOp().create().forPath("/test/bar", "first".getBytes()); Migration migration = () -> Arrays.asList(op1, op2); MigrationSet migrationSet = MigrationSet.build("1", Collections.singletonList(migration)); CountDownLatch latch = new CountDownLatch(1); filterLatch.set(latch); CompletionStage<Void> first = manager.migrate(migrationSet); Assert.assertTrue(timing.awaitLatch(filterIsSetLatch)); MigrationManager manager2 = new MigrationManager(client, LOCK_PATH, META_DATA_PATH, executor, Duration.ofMillis(timing.forSleepingABit().milliseconds())); try { complete(manager2.migrate(migrationSet)); Assert.fail("Should throw"); } catch ( Throwable e ) { Assert.assertTrue(Throwables.getRootCause(e) instanceof AsyncWrappers.TimeoutException, "Should throw AsyncWrappers.TimeoutException, was: " + Throwables.getStackTraceAsString(Throwables.getRootCause(e))); } latch.countDown(); complete(first); Assert.assertEquals(client.unwrap().getData().forPath("/test/bar"), "first".getBytes()); }
@Test public void testConcurrency2() throws Exception { CuratorOp op1 = client.transactionOp().create().forPath("/test"); CuratorOp op2 = client.transactionOp().create().forPath("/test/bar", "first".getBytes()); Migration migration = () -> Arrays.asList(op1, op2); MigrationSet migrationSet = MigrationSet.build("1", Collections.singletonList(migration)); CountDownLatch latch = new CountDownLatch(1); filterLatch.set(latch); CompletionStage<Void> first = manager.migrate(migrationSet); Assert.assertTrue(timing.awaitLatch(filterIsSetLatch)); CompletionStage<Void> second = manager.migrate(migrationSet); try { second.toCompletableFuture().get(timing.forSleepingABit().milliseconds(), TimeUnit.MILLISECONDS); Assert.fail("Should throw"); } catch ( Throwable e ) { Assert.assertTrue(Throwables.getRootCause(e) instanceof TimeoutException, "Should throw TimeoutException, was: " + Throwables.getStackTraceAsString(Throwables.getRootCause(e))); } latch.countDown(); complete(first); Assert.assertEquals(client.unwrap().getData().forPath("/test/bar"), "first".getBytes()); complete(second); Assert.assertEquals(manager.debugCount.get(), 1); }