question: 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 ( ) ) ; } }
target:  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