@Override
		protected void doStop() {
			// stop the offset manager and the channel adapter before unbinding
			// this means that the upstream channel adapter has a chance to stop
			kafkaMessageDrivenChannelAdapter.stop();
			if (messageListenerContainer.getOffsetManager() instanceof DisposableBean) {
				try {
					((DisposableBean) messageListenerContainer.getOffsetManager()).destroy();
				}
				catch (Exception e) {
					logger.error("Error while closing the offset manager", e);
				}
			}
			super.doStop();
		}
Fetches partition data, throwing an exception if the expected number is not met after a number of
subsequent retries
@param topicName topic
@param expectedNumber expected number of partitions. If less or equal to 0, check is suppressed.
@return the partitions
spring-xd-messagebus-kafka/src/main/java/org/springframework/xd/dirt/integration/kafka/KafkaMessageBus.java