@Test
    public void testWrap() throws Exception {
        // Initialize the scheduler & the store.
        int partitionNum = 10;
        IScheduler[] schedulers = new IScheduler[partitionNum];
        for (int i = 0; i < partitionNum; i ++) {
            // We use multiple threads here.
            schedulers[i] = new ExecutorScheduler(Executors.newSingleThreadExecutor());
        }
        final PartitionedSchedulerPool<String> schedulerPool
                = new PartitionedSchedulerPool<String>(schedulers, HashedPartitioner.<String>instance());

        // Initialize the service.
        Client<String, Integer, Boolean> evenChecker = new Client<String, Integer, Boolean>(new Func1<Integer, Boolean>() {
            @Override
            public Boolean apply(Integer value) {
                return value % 2 == 0;
            }
        });

        // Initialize the async client.
        final AsyncClient<String, Integer, Boolean> client = new AsyncClient<String, Integer, Boolean>(evenChecker);

        // Run
        try {
            Waiter<Unit, Maybe<Boolean>>[] waiters = (Waiter<Unit, Maybe<Boolean>>[]) new Waiter<?, ?>[10];
            Random random = new Random();
            // Test when results come back in time.
            for (int i = 0; i < waiters.length; i++) {
                String key = Long.toHexString(random.nextLong());
                Waiter<Unit, Maybe<Boolean>> waiter = client.call(key, i, 100, TimeUnit.MILLISECONDS)
                        .continueWith(new Func1<Maybe<Boolean>, Maybe<Boolean>>() {
                            // Revert the result.
                            @Override
                            public Maybe<Boolean> apply(Maybe<Boolean> value) {
                                return value.isNothing() ? value : Maybe.just(!value.get());
                            }
                        }).continueWithWaiter();
                waiter.execute(schedulerPool.getScheduler(key));
                waiters[i] = waiter;
                Thread.sleep(10);
            }
            for (int i = 0; i < waiters.length; i++) {
                Assert.assertEquals(i % 2 != 0, waiters[i].getResult().get());
            }
            Assert.assertEquals(0, client.getRequestRecords().size());

            // Test when results don't come back in time.
            for (int i = 0; i < waiters.length; i++) {
                String key = Long.toHexString(random.nextLong());
                Waiter<Unit, Maybe<Boolean>> waiter = client.call(key, i, 5, TimeUnit.MILLISECONDS).continueWithWaiter();
                waiter.execute(schedulerPool.getScheduler(key));
                waiters[i] = waiter;
                Thread.sleep(10);
            }
            for (int i = 0; i < waiters.length; i++) {
                Assert.assertTrue(waiters[i].getResult().isNothing());
            }
            Assert.assertEquals(0, client.getRequestRecords().size());

            // Verify that all tasks for the same key are executed in the same thread.
            Thread.sleep(100);
            checkThreadInfo();
        } finally {
            evenChecker.shutdown();
        }
    }