### Question:
PartitionBroker extends BrokerFilter { @Override public void start() throws Exception { super.start(); getExecutor().execute(new Runnable() { @Override public void run() { Thread.currentThread().setName("Partition Monitor"); onMonitorStart(); try { runPartitionMonitor(); } catch (Exception e) { onMonitorStop(); } } }); } PartitionBroker(Broker broker, PartitionBrokerPlugin plugin); @Override void start(); @Override void addConnection(ConnectionContext context, ConnectionInfo info); @Override void removeConnection(ConnectionContext context, ConnectionInfo info, Throwable error); @Override void send(ProducerBrokerExchange producerExchange, Message messageSend);  }### Answer:
@Test(timeout = 1000*60*60) public void testNonFailoverClientHasNoPartitionEffect() throws Exception { partitioning.byClientId = new HashMap<String, Target>(); partitioning.byClientId.put("client1", new Target("broker1")); createBrokerCluster(2); Connection connection = createConnectionToUrl(getConnectURL("broker2")); within(5, TimeUnit.SECONDS, new Task() { public void run() throws Exception { assertEquals(0, getTransportConnector("broker1").getConnections().size()); assertEquals(1, getTransportConnector("broker2").getConnections().size()); } }); connection.setClientID("client1"); connection.start(); Thread.sleep(1000); assertEquals(0, getTransportConnector("broker1").getConnections().size()); assertEquals(1, getTransportConnector("broker2").getConnections().size()); }
@Test(timeout = 1000*60*60) public void testPartitionByClientId() throws Exception { partitioning.byClientId = new HashMap<String, Target>(); partitioning.byClientId.put("client1", new Target("broker1")); partitioning.byClientId.put("client2", new Target("broker2")); createBrokerCluster(2); Connection connection = createConnectionTo("broker2"); within(5, TimeUnit.SECONDS, new Task() { public void run() throws Exception { assertEquals(0, getTransportConnector("broker1").getConnections().size()); assertEquals(1, getTransportConnector("broker2").getConnections().size()); } }); connection.setClientID("client1"); connection.start(); within(5, TimeUnit.SECONDS, new Task() { public void run() throws Exception { assertEquals(1, getTransportConnector("broker1").getConnections().size()); assertEquals(0, getTransportConnector("broker2").getConnections().size()); } }); }