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));
    }
@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));
            fail("Should throw");
        } catch (Throwable e) {
            assertTrue(Throwables.getRootCause(e) instanceof MigrationException);
        }
    }