@Test
	public void testKafkaSpecificConsumerPropertiesAreSet() {
		KafkaTestMessageBus messageBus = (KafkaTestMessageBus) getMessageBus();
		Properties consumerProperties = new Properties();
		consumerProperties.put(KafkaMessageBus.AUTO_COMMIT_OFFSET_ENABLED, "false");
		consumerProperties.put(KafkaMessageBus.FETCH_SIZE, "7");
		consumerProperties.put(KafkaMessageBus.QUEUE_SIZE, "128");
		long uniqueBindingId = System.currentTimeMillis();
		DirectChannel inputChannel = new DirectChannel();
		messageBus.bindConsumer("foo" + uniqueBindingId, inputChannel, consumerProperties);
		DirectFieldAccessor accessor = new DirectFieldAccessor(messageBus.getCoreMessageBus());
		@SuppressWarnings("unchecked")
		List<Binding> bindings = (List<Binding>) accessor.getPropertyValue("bindings");
		assertThat(bindings, hasSize(1));
		Binding consumerBinding = bindings.get(0);
		AbstractEndpoint bindingEndpoint = consumerBinding.getEndpoint();
		DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(bindingEndpoint);
		KafkaMessageListenerContainer messageListenerContainer = ((KafkaMessageListenerContainer) endpointAccessor.getPropertyValue("messageListenerContainer") );
		assertThat(messageListenerContainer.getQueueSize(), equalTo(128));
		assertThat(messageListenerContainer.getMaxFetch(), equalTo(7));
		assertThat(messageListenerContainer.getMessageListener(), instanceOf(AcknowledgingMessageListener.class));
		messageBus.unbindConsumers("foo" + uniqueBindingId);
	}
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