private Collection<Partition> ensureTopicCreated(final String topicName, final int numPartitions,
			int replicationFactor, final boolean createOrResize) {
		// The following is basically copy/paste from AdminUtils.createTopic() with
		// createOrUpdateTopicPartitionAssignmentPathInZK(..., update=true)
		if (createOrResize) {
			final Properties topicConfig = new Properties();
			Seq<Object> brokerList = ZkUtils.getSortedBrokerList(zkClient);
			final scala.collection.Map<Object, Seq<Object>> replicaAssignment =
					AdminUtils.assignReplicasToBrokers(brokerList, numPartitions, replicationFactor, -1, -1);
			retryOperations.execute(new RetryCallback<Object, RuntimeException>() {
				@Override public Object doWithRetry(RetryContext context) throws RuntimeException {
					AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkClient, topicName, replicaAssignment,
							topicConfig, true);
					return null;
				}
			});
		}
		return getPartitions(topicName, numPartitions);
	}
Creates a Kafka topic if needed, or try to increase its partition count to the desired number.
spring-xd-messagebus-kafka/src/main/java/org/springframework/xd/dirt/integration/kafka/KafkaMessageBus.java