@Override
    public void mute(String id) {
        KafkaChannel channel = openOrClosingChannelOrFail(id);
        mute(channel);
    }
@Test
    public void testMute() throws Exception {
        blockingConnect("0");
        blockingConnect("1");

        selector.send(createSend("0", "hello"));
        selector.send(createSend("1", "hi"));

        selector.mute("1");

        while (selector.completedReceives().isEmpty())
            selector.poll(5);
        assertEquals(1, selector.completedReceives().size(), "We should have only one response");
        assertEquals("0", selector.completedReceives().iterator().next().source(),
            "The response should not be from the muted node");

        selector.unmute("1");
        do {
            selector.poll(5);
        } while (selector.completedReceives().isEmpty());
        assertEquals(1, selector.completedReceives().size(), "We should have only one response");
        assertEquals("1", selector.completedReceives().iterator().next().source(), "The response should be from the previously muted node");
    }