@Test
	@SuppressWarnings("unchecked")
	public void testTopicOffsetManagementEnabled() {
		KafkaTestMessageBus bus =
				new KafkaTestMessageBus(kafkaTestSupport, getCodec(),
						KafkaMessageBus.OffsetManagement.kafkaTopic);
		DirectChannel moduleOutputChannel = new DirectChannel();
		QueueChannel moduleInputChannel = new QueueChannel();
		long uniqueBindingId = System.currentTimeMillis();
		bus.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, null);
		bus.bindConsumer("foo" + uniqueBindingId + ".0", moduleInputChannel, null);
		Collection<Binding> bindings = (Collection<Binding>) getBindings(bus);
		assertThat(bindings.size(),equalTo(2));
		for (Binding binding : bindings) {
			if ("consumer".equals(binding.getType())) {
				AbstractEndpoint endpoint = binding.getEndpoint();
				DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(endpoint);
				Object messageListenerContainer = endpointAccessor.getPropertyValue("messageListenerContainer");
				DirectFieldAccessor containerAccessor = new DirectFieldAccessor(messageListenerContainer);
				Object wrapperOffsetManager = containerAccessor.getPropertyValue("offsetManager");
				DirectFieldAccessor offsetManagerAccessor = new DirectFieldAccessor(wrapperOffsetManager);
				Object delegateOffsetManager = offsetManagerAccessor.getPropertyValue("delegate");
				assertThat(delegateOffsetManager, instanceOf(KafkaTopicOffsetManager.class));
			}
		}
		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