private void createKafkaConsumer(String name, final MessageChannel moduleInputChannel, Properties properties,
			boolean pubSub, long referencePoint) {

		// we remove the group qualifier
		String unqualifiedName = pubSub ? BusUtils.removeGroupFromPubSub(name) : name;

		if (pubSub) {
			if (isNamedChannel(unqualifiedName)) {
				validateConsumerProperties(name, properties, SUPPORTED_NAMED_PUBSUB_CONSUMER_PROPERTIES);
			}
			else {
				validateConsumerProperties(name, properties, SUPPORTED_PUBSUB_CONSUMER_PROPERTIES);
			}
		}
		else {
			if (isNamedChannel(unqualifiedName)) {
				validateConsumerProperties(name, properties, SUPPORTED_NAMED_CONSUMER_PROPERTIES);
			}
			else {
				validateConsumerProperties(name, properties, SUPPORTED_CONSUMER_PROPERTIES);
			}
		}
		KafkaPropertiesAccessor accessor = new KafkaPropertiesAccessor(properties);

		int maxConcurrency = accessor.getConcurrency(defaultConcurrency);



		String listenedTopicName = escapeTopicName(unqualifiedName);

		int partitionCountFromConfiguration = accessor.getNumberOfKafkaPartitionsForConsumer();
		Collection<Partition> partitions = initializePartitions(
				unqualifiedName, listenedTopicName, partitionCountFromConfiguration);

		Decoder<byte[]> valueDecoder = new DefaultDecoder(null);
		Decoder<byte[]> keyDecoder = new DefaultDecoder(null);

		Collection<Partition> listenedPartitions;

		int moduleCount = accessor.getCount();

		if (moduleCount == 1) {
			listenedPartitions = partitions;
		}
		else {
			listenedPartitions = new ArrayList<Partition>();
			for (Partition partition : partitions) {
				// divide partitions across modules
				if (accessor.getPartitionIndex() != -1) {
					if ((partition.getId() % moduleCount) == accessor.getPartitionIndex()) {
						listenedPartitions.add(partition);
					}
				}
				else {
					int moduleSequence = accessor.getSequence();
					if (moduleCount == 0) {
						throw new IllegalArgumentException("The Kafka transport does not support 0-count modules");
					}
					else {
						// sequence numbers are zero-based
						if ((partition.getId() % moduleCount) == (moduleSequence - 1)) {
							listenedPartitions.add(partition);
						}
					}
				}
			}
		}

		ReceivingHandler rh = new ReceivingHandler();
		rh.setOutputChannel(moduleInputChannel);

		final FixedSubscriberChannel bridge = new FixedSubscriberChannel(rh);
		bridge.setBeanName("bridge." + name);

		String group = pubSub ? BusUtils.getGroupFromPubSub(name) : POINT_TO_POINT_SEMANTICS_CONSUMER_GROUP;

		boolean resetOffsets = pubSub && !accessor.isDurable(defaultDurableSubscription);

		final KafkaMessageListenerContainer messageListenerContainer =
				createMessageListenerContainer(accessor,
						group, maxConcurrency, listenedPartitions,
						referencePoint, resetOffsets);

		final KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter =
				new KafkaMessageDrivenChannelAdapter(messageListenerContainer);
		kafkaMessageDrivenChannelAdapter.setBeanFactory(this.getBeanFactory());
		kafkaMessageDrivenChannelAdapter.setKeyDecoder(keyDecoder);
		kafkaMessageDrivenChannelAdapter.setPayloadDecoder(valueDecoder);
		kafkaMessageDrivenChannelAdapter.setOutputChannel(bridge);
		kafkaMessageDrivenChannelAdapter.setAutoCommitOffset(
				accessor.getAutoCommitOffsetEnabled(this.defaultAutoCommitOffsetEnabled));
		kafkaMessageDrivenChannelAdapter.afterPropertiesSet();
		kafkaMessageDrivenChannelAdapter.start();


		EventDrivenConsumer edc = new KafkaConsumerBusAdapter(bridge, rh, kafkaMessageDrivenChannelAdapter, messageListenerContainer);
		edc.setBeanName("inbound." + name);

		Binding consumerBinding = Binding.forConsumer(name, edc, moduleInputChannel, accessor);
		addBinding(consumerBinding);
		consumerBinding.start();

	}
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