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;
  }
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
This API is being used by ClusterKafkaSource