public void configure(Context context){
  this.context=context;
  batchUpperLimit=context.getInteger(KafkaSourceConstants.BATCH_SIZE,KafkaSourceConstants.DEFAULT_BATCH_SIZE);
  timeUpperLimit=context.getInteger(KafkaSourceConstants.BATCH_DURATION_MS,KafkaSourceConstants.DEFAULT_BATCH_DURATION);
  topic=context.getString(KafkaSourceConstants.TOPIC);
  if (topic == null) {
    throw new ConfigurationException("Kafka topic must be specified.");
  }
  kafkaProps=KafkaSourceUtil.getKafkaProperties(context);
  consumerTimeout=Integer.parseInt(kafkaProps.getProperty(KafkaSourceConstants.CONSUMER_TIMEOUT));
  kafkaAutoCommitEnabled=Boolean.parseBoolean(kafkaProps.getProperty(KafkaSourceConstants.AUTO_COMMIT_ENABLED));
  if (counter == null) {
    counter=new KafkaSourceCounter(getName());
  }
}

public void configure ( Context context ) { this . context = context ; batchUpperLimit = context . getInteger ( KafkaSourceConstants . BATCH_SIZE , KafkaSourceConstants . DEFAULT_BATCH_SIZE ) ; timeUpperLimit = context . getInteger ( KafkaSourceConstants . BATCH_DURATION_MS , KafkaSourceConstants . DEFAULT_BATCH_DURATION ) ; topic = context . getString ( KafkaSourceConstants . TOPIC ) ; if ( topic == null ) { throw new ConfigurationException ( "Kafka topic must be specified." ) ; } kafkaProps = KafkaSourceUtil . getKafkaProperties ( context ) ; consumerTimeout = Integer . parseInt ( kafkaProps . getProperty ( KafkaSourceConstants . CONSUMER_TIMEOUT ) ) ; kafkaAutoCommitEnabled = Boolean . parseBoolean ( kafkaProps . getProperty ( KafkaSourceConstants . AUTO_COMMIT_ENABLED ) ) ; if ( counter == null ) { counter = new KafkaSourceCounter ( getName ( ) ) ; } }
we configure the source and generate properties for the kafka consumer kafka consumer properties are generated as follows : 1 . generate a properties object with some static defaults that can be overridden by source configuration 2 . we add the configuration users added for kafka ( parameters starting with kafka . and must be valid kafka consumer properties 3 . we add the source documented parameters which can override other properties
