@Test
	@SuppressWarnings("unchecked")
	public void testSendAndReceivePubSubWithMultipleConsumers() throws Exception {
		MessageBus messageBus = getMessageBus();
		DirectChannel graultUpstreamModuleOutputChannel = new DirectChannel();
		// Test pub/sub by emulating how StreamPlugin handles taps
		DirectChannel tapChannel = new DirectChannel();
		QueueChannel graultDownstreamModuleInputChannel = new QueueChannel();
		QueueChannel fooTapDownstreamInputChannel = new QueueChannel();
		QueueChannel barTapDowstreamInput1Channel = new QueueChannel();
		QueueChannel barTapDowstreamInput2Channel = new QueueChannel();
		String originalTopic = "grault.0";
		messageBus.bindProducer(originalTopic, graultUpstreamModuleOutputChannel, null);
		messageBus.bindConsumer(originalTopic, graultDownstreamModuleInputChannel, null);
		graultUpstreamModuleOutputChannel.addInterceptor(new WireTap(tapChannel));
		Properties graultTapProducerProperties = new Properties();
		// set the partition count to two, so that the tap has two partitions as well
		// this will be necessary to robin messages across the competing consumers
		graultTapProducerProperties.setProperty(BusProperties.MIN_PARTITION_COUNT, "2");
		messageBus.bindPubSubProducer("tap:grault.http", tapChannel, graultTapProducerProperties);
		// A new module is using the tap as an input channel
		String fooTapName = messageBus.isCapable(MessageBus.Capability.DURABLE_PUBSUB) ? "foo.tap:grault.http" : "tap:grault.http";
		messageBus.bindPubSubConsumer(fooTapName, fooTapDownstreamInputChannel, null);
		// Another new module is using tap as an input channel
		String barTapName = messageBus.isCapable(MessageBus.Capability.DURABLE_PUBSUB) ? "bar.tap:grault.http" : "tap:grault.http";
		Properties barTap1Properties = new Properties();
		barTap1Properties.setProperty(BusProperties.COUNT, "2");
		barTap1Properties.setProperty(BusProperties.SEQUENCE, "1");
		messageBus.bindPubSubConsumer(barTapName, barTapDowstreamInput1Channel, barTap1Properties);
		Properties barTap2Properties = new Properties();
		barTap2Properties.setProperty(BusProperties.COUNT, "2");
		barTap2Properties.setProperty(BusProperties.SEQUENCE, "2");
		messageBus.bindPubSubConsumer(barTapName, barTapDowstreamInput2Channel, barTap2Properties);
		Message<?> message1 = MessageBuilder.withPayload("foo1").setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar")
				.build();
		Message<?> message2 = MessageBuilder.withPayload("foo2").setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar")
				.build();

		graultUpstreamModuleOutputChannel.send(message1);
		graultUpstreamModuleOutputChannel.send(message2);
		Message<?> graultDownstreamInbound = graultDownstreamModuleInputChannel.receive(5000);
		assertNotNull(graultDownstreamInbound);
		assertEquals("foo1", graultDownstreamInbound.getPayload());
		assertNull(graultDownstreamInbound.getHeaders().get(XdHeaders.XD_ORIGINAL_CONTENT_TYPE));
		assertEquals("foo/bar", graultDownstreamInbound.getHeaders().get(MessageHeaders.CONTENT_TYPE));
		graultDownstreamInbound = graultDownstreamModuleInputChannel.receive(5000);
		assertNotNull(graultDownstreamInbound);
		assertEquals("foo2", graultDownstreamInbound.getPayload());
		assertNull(graultDownstreamInbound.getHeaders().get(XdHeaders.XD_ORIGINAL_CONTENT_TYPE));
		assertEquals("foo/bar", graultDownstreamInbound.getHeaders().get(MessageHeaders.CONTENT_TYPE));
		List<Message<?>> tappedFoo = new ArrayList<>();
		tappedFoo.add(fooTapDownstreamInputChannel.receive(5000));
		tappedFoo.add(fooTapDownstreamInputChannel.receive(5000));
		Message<?> tappedBar1 = barTapDowstreamInput1Channel.receive(5000);
		Message<?> tappedBar2 = barTapDowstreamInput2Channel.receive(5000);
		assertThat(tappedFoo,
				CoreMatchers.<Message<?>>hasItems(hasProperty("payload", equalTo("foo1")),
						hasProperty("payload", equalTo("foo2"))));
		assertThat(tappedFoo,
				everyItem(HasPropertyWithValue.<Message<?>>hasProperty("headers",
						hasEntry(XdHeaders.XD_ORIGINAL_CONTENT_TYPE, null))));
		assertThat(tappedFoo,
				everyItem(HasPropertyWithValue.<Message<?>>hasProperty("headers",
						hasEntry(MessageHeaders.CONTENT_TYPE, "foo/bar"))));
		assertEquals("foo1", tappedBar1.getPayload());
		assertNull(tappedBar1.getHeaders().get(XdHeaders.XD_ORIGINAL_CONTENT_TYPE));
		assertEquals("foo/bar", tappedBar1.getHeaders().get(MessageHeaders.CONTENT_TYPE));
		assertEquals("foo2", tappedBar2.getPayload());
		assertNull(tappedBar2.getHeaders().get(XdHeaders.XD_ORIGINAL_CONTENT_TYPE));
		assertEquals("foo/bar", tappedBar2.getHeaders().get(MessageHeaders.CONTENT_TYPE));

		messageBus.unbindConsumers(fooTapName);
		messageBus.unbindConsumers(originalTopic);
		messageBus.unbindProducers(originalTopic);
		messageBus.unbindProducers("tap:grault.http");
		messageBus.unbindConsumers(barTapName);
		assertTrue(getBindings(messageBus).isEmpty());
	}
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