private KafkaMessageListenerContainer createMessageListenerContainer(KafkaPropertiesAccessor accessor,
			String group, int maxConcurrency, String topic, Collection<Partition> listenedPartitions,
			long referencePoint, boolean resetOffsets) {
		Assert.isTrue(StringUtils.hasText(topic) ^ !CollectionUtils.isEmpty(listenedPartitions),
				"Exactly one of topic or a list of listened partitions must be provided");
		KafkaMessageListenerContainer messageListenerContainer;
		if (topic != null) {
			messageListenerContainer = new KafkaMessageListenerContainer(connectionFactory, topic);
		}
		else {
			messageListenerContainer = new KafkaMessageListenerContainer(connectionFactory,
					listenedPartitions.toArray(new Partition[listenedPartitions.size()]));
		}
		if (logger.isDebugEnabled()) {
			logger.debug("Listening to topic " + topic);
		}
		// if we have less target partitions than target concurrency, adjust accordingly
		messageListenerContainer.setConcurrency(Math.min(maxConcurrency, listenedPartitions.size()));
		OffsetManager offsetManager = createOffsetManager(group, referencePoint);
		if (resetOffsets) {
			offsetManager.resetOffsets(listenedPartitions);
		}
		messageListenerContainer.setOffsetManager(offsetManager);
		int queueSize = accessor.getProperty(QUEUE_SIZE, defaultQueueSize);
		Assert.isTrue(queueSize > 0 && Integer.bitCount(queueSize) == 1, "must be a power of 2");
		messageListenerContainer.setQueueSize(queueSize);
		messageListenerContainer.setMaxFetch(accessor.getProperty(FETCH_SIZE, defaultFetchSize));
		return messageListenerContainer;
	}
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