@Test
	@Override
	@SuppressWarnings("unchecked")
	public void testPartitioningWithSingleReceiver() throws Exception {
		MessageBus bus = getMessageBus();
		String topicName = "foo" + System.currentTimeMillis() + ".0";
		try {
			int partitionCount = 4;
			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.MIN_PARTITION_COUNT, Integer.toString(partitionCount));
			DirectChannel moduleOutputChannel = new DirectChannel();
			QueueChannel moduleInputChannel = new QueueChannel();
			bus.bindProducer(topicName, moduleOutputChannel, properties);
			bus.bindConsumer(topicName, moduleInputChannel, null);
			int totalSent = 0;
			for (int i = 0; i < partitionCount; i++) {
				for (int j = 0; j < partitionCount + 1; j++ ) {
					// the distribution is uneven across partitions, so that we can verify that the bus doesn't round robin
					moduleOutputChannel.send(new GenericMessage<Object>(i*partitionCount + j));
					totalSent++;
				}
			}
			List<Message<Integer>> receivedMessages = new ArrayList<>();
			for (int i = 0; i < totalSent; i++) {
				assertTrue(receivedMessages.add((Message<Integer>) moduleInputChannel.receive(2000)));
			}

			assertThat(receivedMessages, hasSize(totalSent));
			for (Message<Integer> receivedMessage : receivedMessages) {
				int expectedPartition = receivedMessage.getPayload() % partitionCount;
				assertThat(expectedPartition, equalTo(receivedMessage.getHeaders().get(KafkaHeaders.PARTITION_ID)));
			}
		}
		finally {
			bus.unbindConsumers(topicName);
			bus.unbindProducers(topicName);
			AdminUtils$.MODULE$.deleteTopic(kafkaTestSupport.getZkClient(),topicName);
		}
	}
Integration tests for the {@link KafkaMessageBus}.

@author Eric Bottard
@author Marius Bogoevici
spring-xd-dirt/src/test/java/org/springframework/xd/dirt/integration/bus/kafka/KafkaMessageBusTests.java