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 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);
        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));
            fail("Should throw");
        } catch (Throwable e) {
            assertTrue(
                    Throwables.getRootCause(e) instanceof AsyncWrappers.TimeoutException,
                    "Should throw AsyncWrappers.TimeoutException, was: "
                            + Throwables.getStackTraceAsString(Throwables.getRootCause(e)));
        }

        latch.countDown();
        complete(first);
        assertArrayEquals(client.unwrap().getData().forPath("/test/bar"), "first".getBytes());
    }