code: private void prepareTopic(final Map<String, InternalTopicMetadata> topicPartitions) {
      log.debug("Starting to validate internal topics {} in partition assignor.", topicPartitions);
  
      // first construct the topics to make ready
      final Map<String, InternalTopicConfig> topicsToMakeReady = new HashMap<>();
  
      for (final InternalTopicMetadata metadata : topicPartitions.values()) {
        final InternalTopicConfig topic = metadata.config;
        final int numPartitions = metadata.numPartitions;
  
        if (numPartitions < 0) {
          throw new StreamsException(String.format("%sTopic [%s] number of partitions not defined", logPrefix, topic.name()));
        }
  
        topic.setNumberOfPartitions(numPartitions);
        topicsToMakeReady.put(topic.name(), topic);
      }
  
      if (!topicsToMakeReady.isEmpty()) {
        internalTopicManager.makeReady(topicsToMakeReady);
      }
  
      log.debug("Completed validating internal topics {} in partition assignor.", topicPartitions);
    }
docs: Internal helper function that creates a Kafka topic
  
  @param topicPartitions Map that contains the topic names to be created with the number of partitions