kafka_source-protolib/src/main/java/com/streamsets/pipeline/stage/origin/kafka/BaseKafkaSource.java
public int getParallelism() throws StageException {
    if (originParallelism == 0) {
      //origin parallelism is not yet calculated
      originParallelism = kafkaValidationUtil.getPartitionCount(
          conf.metadataBrokerList,
          conf.topic,
          new HashMap<String, Object>(conf.kafkaConsumerConfigs),
          3,
          1000
      );
      if(originParallelism < 1) {
        throw new StageException(KafkaErrors.KAFKA_42, conf.topic);
      }
    }
    return originParallelism;
  }
public int getParallelism() throws StageException {
    if (originParallelism == 0) {
      //origin parallelism is not yet calculated
      originParallelism = kafkaValidationUtil.getPartitionCount(
          conf.metadataBrokerList,
          conf.topic,
          new HashMap<String, Object>(conf.kafkaConsumerConfigs),
          3,
          1000
      );
      if(originParallelism < 1) {
        throw new StageException(KafkaErrors.KAFKA_42, conf.topic);
      }
    }
    return originParallelism;
  }
This API is being used by ClusterKafkaSource