@Test
	public void testPartitionedModuleJava() throws Exception {
		MessageBus bus = getMessageBus();
		Properties properties = new Properties();
		properties.put("partitionKeyExtractorClass", "org.springframework.xd.dirt.integration.bus.PartitionTestSupport");
		properties.put("partitionSelectorClass", "org.springframework.xd.dirt.integration.bus.PartitionTestSupport");
		properties.put(BusProperties.NEXT_MODULE_COUNT, "3");
		properties.put(BusProperties.NEXT_MODULE_CONCURRENCY, "2");

		DirectChannel output = new DirectChannel();
		output.setBeanName("test.output");
		bus.bindProducer("partJ.0", output, properties);
		@SuppressWarnings("unchecked")
		List<Binding> bindings = TestUtils.getPropertyValue(bus, "messageBus.bindings", List.class);
		assertEquals(1, bindings.size());
		if (usesExplicitRouting()) {
			AbstractEndpoint endpoint = bindings.get(0).getEndpoint();
			assertThat(getEndpointRouting(endpoint), containsString("partJ.0-' + headers['partition']"));
		}

		properties.clear();
		properties.put("concurrency", "2");
		properties.put("count","3");
		properties.put("partitionIndex", "0");
		QueueChannel input0 = new QueueChannel();
		input0.setBeanName("test.input0J");
		bus.bindConsumer("partJ.0", input0, properties);
		properties.put("partitionIndex", "1");
		QueueChannel input1 = new QueueChannel();
		input1.setBeanName("test.input1J");
		bus.bindConsumer("partJ.0", input1, properties);
		properties.put("partitionIndex", "2");
		QueueChannel input2 = new QueueChannel();
		input2.setBeanName("test.input2J");
		bus.bindConsumer("partJ.0", input2, properties);

		output.send(new GenericMessage<Integer>(2));
		output.send(new GenericMessage<Integer>(1));
		output.send(new GenericMessage<Integer>(0));

		Message<?> receive0 = input0.receive(1000);
		assertNotNull(receive0);
		Message<?> receive1 = input1.receive(1000);
		assertNotNull(receive1);
		Message<?> receive2 = input2.receive(1000);
		assertNotNull(receive2);

		if (usesExplicitRouting()) {
			assertEquals(0, receive0.getPayload());
			assertEquals(1, receive1.getPayload());
			assertEquals(2, receive2.getPayload());
		}
		else {

			assertThat(Arrays.asList(
					(Integer) receive0.getPayload(),
					(Integer) receive1.getPayload(),
					(Integer) receive2.getPayload()),
					containsInAnyOrder(0, 1, 2));
		}

		bus.unbindConsumers("partJ.0");
		bus.unbindProducers("partJ.0");
	}
Tests for buses that support partitioning.

@author Gary Russell
spring-xd-dirt/src/test/java/org/springframework/xd/dirt/integration/bus/PartitionCapableBusTests.java