private void testPartitioning(Transport transport) throws Exception {
		Properties systemProperties = new Properties();
		systemProperties.setProperty("xd.transport", transport.toString());

		for (int i = 0; i < 2; i++) {
			startContainer(systemProperties);
		}

		SpringXDTemplate template = ensureTemplate();
		logger.info("Waiting for containers...");
		waitForContainers();
		logger.info("Containers running");

		String streamName = testName.getMethodName() + "-woodchuck";
		File file = File.createTempFile("temp", ".txt");
		file.deleteOnExit();
		int httpPort = SocketUtils.findAvailableServerSocket();
		template.streamOperations().createStream(
				streamName,
				String.format("http --port=%s | splitter --expression=payload.split(' ') | " +
						"file --dir=%s --name=${xd.container.id}", httpPort, file.getParent()), false);
		verifyStreamCreated(streamName);

		template.streamOperations().deploy(streamName, DeploymentPropertiesFormat.parseDeploymentProperties(
				"module.splitter.producer.partitionKeyExpression=payload,module.file.count=2"));

		// verify modules
		Map<String, Properties> modules = new HashMap<String, Properties>();
		int attempts = 0;
		while (attempts++ < 60) {
			Thread.sleep(500);
			for (ModuleMetadataResource module : template.runtimeOperations().listDeployedModules()) {
				modules.put(module.getContainerId() + ":" + module.getModuleType() + ":" + module.getName(),
						module.getDeploymentProperties());
			}
			if (modules.size() == 4) {
				break;
			}
		}
		assertEquals("timed out waiting for stream modules to deploy", 4, modules.size());

		RestTemplate restTemplate = new RestTemplate();
		String text = "how much wood would a woodchuck chuck if a woodchuck could chuck wood";
		int postAttempts = 0;
		while (postAttempts++ < 50) {
			Thread.sleep(100);
			try {
				ResponseEntity<?> entity = restTemplate.postForEntity("http://localhost:" + httpPort, text,
						String.class);
				assertEquals(HttpStatus.OK, entity.getStatusCode());
				break;
			}
			catch (Exception e) {
				// will try again
			}
		}

		File[] outputFiles = new File[2];
		for (Map.Entry<String, Properties> module : modules.entrySet()) {
			if (module.getKey().contains("sink")) {
				int index = Integer.parseInt(module.getValue().getProperty("consumer.partitionIndex"));
				String container = module.getKey().substring(0, module.getKey().indexOf(':'));
				File output = new File(file.getParent(), container + ".out");
				output.deleteOnExit();
				outputFiles[index] = output;
			}
		}

		String expectedPartition0 = "how\nchuck\nchuck\n";
		String expectedPartition1 = "much\nwood\nwould\na\nwoodchuck\nif\na\nwoodchuck\ncould\nwood\n";

		int fileAttempts = 0;
		String[] results = new String[2];
		while (fileAttempts++ < 10) {
			Thread.sleep(500);
			if (outputFiles[0].exists() && outputFiles[1].exists()) {
				results[0] = FileCopyUtils.copyToString(new FileReader(outputFiles[0]));
				results[1] = FileCopyUtils.copyToString(new FileReader(outputFiles[1]));
				if (results[0].length() == expectedPartition0.length()
						&& results[1].length() == expectedPartition1.length()) {
					break;
				}
			}
		}
		assertEquals(expectedPartition0, results[0]);
		assertEquals(expectedPartition1, results[1]);

	}
Test data partitioning for the given transport. This method
asserts that data sent through a source is correctly routed
to the expected sinks based on the partitioning configuration.

@param transport transport to use for partition testing
@throws Exception
spring-xd-distributed-test/src/test/java/org/springframework/xd/distributed/test/StreamPartitionTests.java