### Question:
PartitionBroker extends BrokerFilter { @Override public void send(ProducerBrokerExchange producerExchange, Message messageSend) throws Exception { ConnectionMonitor monitor = monitors.get(producerExchange.getConnectionContext().getConnectionId()); if( monitor!=null ) { monitor.onSend(producerExchange, messageSend); } } 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 testPartitionByQueue() throws Exception { partitioning.byQueue = new HashMap<String, Target>(); partitioning.byQueue.put("foo", new Target("broker1")); createBrokerCluster(2); Connection connection2 = 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()); } }); Session session2 = connection2.createSession(false, Session.AUTO_ACKNOWLEDGE); MessageConsumer consumer = session2.createConsumer(session2.createQueue("foo")); within(5, TimeUnit.SECONDS, new Task() { public void run() throws Exception { assertEquals(1, getTransportConnector("broker1").getConnections().size()); assertEquals(0, getTransportConnector("broker2").getConnections().size()); } }); Connection connection1 = createConnectionTo("broker2"); Session session1 = connection1.createSession(false, Session.AUTO_ACKNOWLEDGE); MessageProducer producer = session1.createProducer(session1.createQueue("foo")); within(5, TimeUnit.SECONDS, new Task() { public void run() throws Exception { assertEquals(1, getTransportConnector("broker1").getConnections().size()); assertEquals(1, getTransportConnector("broker2").getConnections().size()); } }); for (int i = 0; i < 100; i++) { producer.send(session1.createTextMessage("#" + i)); } within(5, TimeUnit.SECONDS, new Task() { public void run() throws Exception { assertEquals(2, getTransportConnector("broker1").getConnections().size()); assertEquals(0, getTransportConnector("broker2").getConnections().size()); } }); }