@Test
	public void testMoreHeaders() throws Exception {
		KafkaTestMessageBus bus =
				new KafkaTestMessageBus(kafkaTestSupport, getCodec(), KafkaMessageBus.Mode.embeddedHeaders,
						"propagatedHeader");
		Collection<String> headers = Arrays.asList(TestUtils.getPropertyValue(bus.getCoreMessageBus(), "headersToMap",
				String[].class));
		assertTrue(headers.contains("propagatedHeader"));
		assertEquals(XdHeaders.STANDARD_HEADERS.length + 1, headers.size());
		DirectChannel moduleOutputChannel = new DirectChannel();
		QueueChannel moduleInputChannel = new QueueChannel();
		long uniqueBindingId = System.currentTimeMillis();
		Properties emptyProperties = new Properties();
		bus.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, emptyProperties);
		bus.bindConsumer("foo" + uniqueBindingId + ".0", moduleInputChannel, emptyProperties);
		Message<?> message = org.springframework.integration.support.MessageBuilder.
				withPayload("payload").setHeader("propagatedHeader","propagatedValue").build();
		// Let the consumer actually bind to the producer before sending a msg
		busBindUnbindLatency();
		moduleOutputChannel.send(message);
		Message<?> inbound = moduleInputChannel.receive(2000);
		assertThat(inbound.getHeaders(), IsMapContaining.hasKey("propagatedHeader"));
		assertThat((String)inbound.getHeaders().get("propagatedHeader"), equalTo("propagatedValue"));
		bus.unbindProducers("foo" + uniqueBindingId + ".0");
		bus.unbindConsumers("foo" + uniqueBindingId + ".0");
		bus.cleanup();
	}
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