    @Test
    public void testGetOwnerMulti() throws Exception {

        ServerConfiguration cfg1 = new CustomServerConfiguration(cfg.getServerPort() + 1), cfg2 = new CustomServerConfiguration(
            cfg.getServerPort() + 2);
        // TODO change cfg1 cfg2 params
        ZkTopicManager tm1 = new ZkTopicManager(zk, cfg1, scheduler),
                       tm2 = new ZkTopicManager(zk, cfg2, scheduler);

        tm.getOwner(topic, false, addrCbq, null);
        HedwigSocketAddress owner = check(addrCbq.take());

        // If we were told to have another person claim the topic, make them
        // claim the topic.
        if (owner.getPort() == cfg1.getServerPort())
            tm1.getOwner(topic, true, addrCbq, null);
        else if (owner.getPort() == cfg2.getServerPort())
            tm2.getOwner(topic, true, addrCbq, null);
        if (owner.getPort() != cfg.getServerPort())
            Assert.assertEquals(owner, check(addrCbq.take()));

        for (int i = 0; i < 100; ++i) {
            tm.getOwner(topic, false, addrCbq, null);
            Assert.assertEquals(owner, check(addrCbq.take()));

            tm1.getOwner(topic, false, addrCbq, null);
            Assert.assertEquals(owner, check(addrCbq.take()));

            tm2.getOwner(topic, false, addrCbq, null);
            Assert.assertEquals(owner, check(addrCbq.take()));
        }

        // Give us 100 chances to choose another owner if not shouldClaim.
        for (int i = 0; i < 100; ++i) {
            if (!owner.equals(me))
                break;
            tm.getOwner(mkTopic(i), false, addrCbq, null);
            owner = check(addrCbq.take());
            if (i == 99)
                Assert.fail("Never chose another owner");
        }

        // Make sure we always choose ourselves if shouldClaim.
        for (int i = 0; i < 100; ++i) {
            tm.getOwner(mkTopic(100), true, addrCbq, null);
            Assert.assertEquals(me, check(addrCbq.take()));
        }

        tm1.stop();
        tm2.stop();
    }