@Test
	public void testCustomPartitionCountDoesNotOverrideModuleCountAndConcurrencyIfSmaller() throws Exception {

		byte[] ratherBigPayload = new byte[2048];
		Arrays.fill(ratherBigPayload, (byte) 65);
		KafkaTestMessageBus messageBus = (KafkaTestMessageBus) getMessageBus();


		DirectChannel moduleOutputChannel = new DirectChannel();
		QueueChannel moduleInputChannel = new QueueChannel();
		Properties producerProps = new Properties();
		producerProps.put(BusProperties.MIN_PARTITION_COUNT, "5");
		producerProps.put(BusProperties.NEXT_MODULE_CONCURRENCY, "6");
		Properties consumerProps = new Properties();
		consumerProps.put(BusProperties.MIN_PARTITION_COUNT, "5");
		consumerProps.put(BusProperties.CONCURRENCY, "6");
		long uniqueBindingId = System.currentTimeMillis();
		messageBus.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, producerProps);
		messageBus.bindConsumer("foo" + uniqueBindingId + ".0", moduleInputChannel, consumerProps);
		Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload(ratherBigPayload).build();
		// Let the consumer actually bind to the producer before sending a msg
		busBindUnbindLatency();
		moduleOutputChannel.send(message);
		Message<?> inbound = moduleInputChannel.receive(2000);
		assertNotNull(inbound);
		assertArrayEquals(ratherBigPayload, (byte[]) inbound.getPayload());
		Collection<Partition> partitions = messageBus.getCoreMessageBus().getConnectionFactory().getPartitions(
				"foo" + uniqueBindingId + ".0");
		assertThat(partitions, hasSize(6));
		messageBus.unbindProducers("foo" + uniqueBindingId + ".0");
		messageBus.unbindConsumers("foo" + uniqueBindingId + ".0");
	}
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