code: public ILog getOrCreateLog(String topic, int partition) throws IOException {
          final int configPartitionNumber = getPartition(topic);
          if (partition >= configPartitionNumber) {
              throw new IOException("partition is bigger than the number of configuration: " + configPartitionNumber);
          }
          boolean hasNewTopic = false;
          Pool<Integer, Log> parts = getLogPool(topic, partition);
          if (parts == null) {
              Pool<Integer, Log> found = logs.putIfNotExists(topic, new Pool<Integer, Log>());
              if (found == null) {
                  hasNewTopic = true;
              }
              parts = logs.get(topic);
          }
          //
          Log log = parts.get(partition);
          if (log == null) {
              log = createLog(topic, partition);
              Log found = parts.putIfNotExists(partition, log);
              if (found != null) {
                  Closer.closeQuietly(log, logger);
                  log = found;
              } else {
                  logger.info(format("Created log for [%s-%d], now create other logs if necessary", topic, partition));
                  final int configPartitions = getPartition(topic);
                  for (int i = 0; i < configPartitions; i++) {
                      getOrCreateLog(topic, i);
                  }
              }
          }
          if (hasNewTopic && config.getEnableZookeeper()) {
              topicRegisterTasks.add(new TopicTask(TopicTask.TaskType.CREATE, topic));
          }
          return log;
      }
docs: Create the log if it does not exist or return back exist log
  
  @param topic     the topic name
  @param partition the partition id
  @return read or create a log
  @throws IOException any IOException