private OffsetManager createOffsetManager(String group, long referencePoint) {
		try {
			OffsetManager delegateOffsetManager;
			if (OffsetManagement.kafkaNative.equals(offsetManagement)) {
				KafkaNativeOffsetManager kafkaOffsetManager = new KafkaNativeOffsetManager(connectionFactory,
						zookeeperConnect, Collections.<Partition, Long>emptyMap());
				kafkaOffsetManager.setConsumerId(group);
				kafkaOffsetManager.setReferenceTimestamp(referencePoint);
				kafkaOffsetManager.afterPropertiesSet();
				delegateOffsetManager = kafkaOffsetManager;
			}
			else {
				KafkaTopicOffsetManager kafkaOffsetManager = new KafkaTopicOffsetManager(zookeeperConnect,
						offsetStoreTopic, Collections.<Partition, Long>emptyMap());
				kafkaOffsetManager.setConsumerId(group);
				kafkaOffsetManager.setReferenceTimestamp(referencePoint);
				kafkaOffsetManager.setSegmentSize(offsetStoreSegmentSize);
				kafkaOffsetManager.setRetentionTime(offsetStoreRetentionTime);
				kafkaOffsetManager.setRequiredAcks(offsetStoreRequiredAcks);
				kafkaOffsetManager.setMaxSize(offsetStoreMaxFetchSize);
				kafkaOffsetManager.setBatchBytes(offsetStoreBatchBytes);
				kafkaOffsetManager.setMaxQueueBufferingTime(offsetStoreBatchTime);
				kafkaOffsetManager.afterPropertiesSet();
				delegateOffsetManager = kafkaOffsetManager;
			}
			WindowingOffsetManager windowingOffsetManager = new WindowingOffsetManager(delegateOffsetManager);
			windowingOffsetManager.setTimespan(offsetUpdateTimeWindow);
			windowingOffsetManager.setCount(offsetUpdateCount);
			windowingOffsetManager.setShutdownTimeout(offsetUpdateShutdownTimeout);

			windowingOffsetManager.afterPropertiesSet();
			return windowingOffsetManager;
		}
		catch (Exception e) {
			throw new RuntimeException(e);
		}
	}
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