@Test
	public void testKafkaSpecificProducerPropertiesAreSet() {
		KafkaTestMessageBus messageBus = (KafkaTestMessageBus) getMessageBus();
		Properties producerProperties = new Properties();
		producerProperties.put(BusProperties.BATCH_SIZE, "34");
		producerProperties.put(BusProperties.BATCH_TIMEOUT, "7");
		long uniqueBindingId = System.currentTimeMillis();
		DirectChannel inputChannel = new DirectChannel();
		messageBus.bindProducer("foo" + uniqueBindingId, inputChannel, producerProperties);
		DirectFieldAccessor accessor = new DirectFieldAccessor(messageBus.getCoreMessageBus());
		@SuppressWarnings("unchecked")
		List<Binding> bindings = (List<Binding>) accessor.getPropertyValue("bindings");
		assertThat(bindings, hasSize(1));
		Binding producerBinding = bindings.get(0);
		AbstractEndpoint bindingEndpoint = producerBinding.getEndpoint();
		DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(bindingEndpoint);
		MessageHandler messageHandler = ((MessageHandler) endpointAccessor.getPropertyValue("handler") );
		DirectFieldAccessor messageHandlerAccessor = new DirectFieldAccessor(messageHandler);
		ProducerConfiguration<?,?> producerConfiguration = (ProducerConfiguration<?, ?>) messageHandlerAccessor.getPropertyValue("producerConfiguration");
		DirectFieldAccessor producerConfigurationAccessor = new DirectFieldAccessor(producerConfiguration);
		@SuppressWarnings("rawtypes")
		KafkaProducer producer = (KafkaProducer) producerConfigurationAccessor.getPropertyValue("producer");
		DirectFieldAccessor producerAccessor = new DirectFieldAccessor(producer);
		ProducerConfig producerConfig = (ProducerConfig)producerAccessor.getPropertyValue("producerConfig");
		assertThat(producerConfig.getInt(ProducerConfig.BATCH_SIZE_CONFIG), equalTo(34));
		assertThat(producerConfig.getLong(ProducerConfig.LINGER_MS_CONFIG), equalTo(7L));
		messageBus.unbindProducers("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