@Test
	public void partitionedJob() {
		String moduleGroupName = "partitionedJob";
		int moduleIndex = 0;
		Module module = Mockito.mock(Module.class);
		when(module.getType()).thenReturn(ModuleType.job);
		Properties properties = new Properties();
		when(module.getProperties()).thenReturn(properties);
		ModuleDeploymentProperties deploymentProperties = new ModuleDeploymentProperties();
		deploymentProperties.put("consumer.concurrency", "2");
		when(module.getDeploymentProperties()).thenReturn(deploymentProperties);
		when(module.getDescriptor()).thenReturn(
				new ModuleDescriptor.Builder().setGroup(moduleGroupName).setIndex(moduleIndex).setModuleDefinition(
						TestModuleDefinitions.dummy("testjob", ModuleType.job)).build());

		MessageChannel stepsOut = new DirectChannel();
		when(module.getComponent("stepExecutionRequests.output", MessageChannel.class)).thenReturn(stepsOut);
		PollableChannel stepResultsIn = new QueueChannel();
		when(module.getComponent("stepExecutionReplies.input", MessageChannel.class)).thenReturn(stepResultsIn);
		final PollableChannel stepsIn = new QueueChannel();
		DirectChannel boundStepsIn = new DirectChannel();
		boundStepsIn.subscribe(new MessageHandler() {

			@Override
			public void handleMessage(Message<?> message) throws MessagingException {
				try {
					Thread.sleep(100);
				}
				catch (InterruptedException e) {
					Thread.currentThread().interrupt();
				}
				stepsIn.send(MessageBuilder.fromMessage(message)
						.setHeader("thread", Thread.currentThread().getName())
						.build());
			}
		});
		when(module.getComponent("stepExecutionRequests.input", MessageChannel.class)).thenReturn(boundStepsIn);
		MessageChannel stepResultsOut = new DirectChannel();
		when(module.getComponent("stepExecutionReplies.output", MessageChannel.class)).thenReturn(stepResultsOut);
		jobPartitionerPlugin.preProcessModule(module);
		jobPartitionerPlugin.postProcessModule(module);
		checkBusBound(messageBus);
		assertEquals(1, TestUtils.getPropertyValue(messageBus, "reqRepExecutors", Map.class).size());
		stepsOut.send(new GenericMessage<String>("foo1"));
		stepsOut.send(new GenericMessage<String>("foo2"));
		Message<?> stepExecutionRequest1 = stepsIn.receive(10000);
		Message<?> stepExecutionRequest2 = stepsIn.receive(10000);
		assertThat(stepExecutionRequest1, anyOf(hasPayload("foo1"), hasPayload("foo2")));
		assertThat(stepExecutionRequest2, anyOf(hasPayload("foo1"), hasPayload("foo2")));
		stepResultsOut.send(MessageBuilder.withPayload("bar1")
				.copyHeaders(stepExecutionRequest1.getHeaders()) // replyTo
				.build());
		stepResultsOut.send(MessageBuilder.withPayload("bar2")
				.copyHeaders(stepExecutionRequest1.getHeaders()) // replyTo
				.build());
		assertThat(stepResultsIn.receive(10000), anyOf(hasPayload("bar1"), hasPayload("bar2")));
		assertThat(stepResultsIn.receive(10000), anyOf(hasPayload("bar1"), hasPayload("bar2")));
		Set<String> threads = new HashSet<String>();
		threads.add((String) stepExecutionRequest1.getHeaders().get("thread"));
		threads.add((String) stepExecutionRequest2.getHeaders().get("thread"));
		assertEquals(2, threads.size());
		jobPartitionerPlugin.removeModule(module);
		checkBusUnbound(messageBus);
		assertEquals(0, TestUtils.getPropertyValue(messageBus, "reqRepExecutors", Map.class).size());
	}
@author Michael Minella
@author Gunnar Hillert
@author Gary Russell
@author Ilayaperumal Gopinathan
@author David Turanski
spring-xd-dirt/src/test/java/org/springframework/xd/dirt/plugins/job/JobPluginTests.java