class KafkaSource extends AbstractPollableSource implements Configurable, BatchSizeSupported {

  @Override
  protected void doConfigure(Context context) throws FlumeException {
    this.context = context;
    headers = new HashMap<String, String>(4);
    tpAndOffsetMetadata = new HashMap<TopicPartition, OffsetAndMetadata>();
    rebalanceFlag = new AtomicBoolean(false);
    kafkaProps = new Properties();

    // can be removed in the next release
    // See https://issues.apache.org/jira/browse/FLUME-2896
    translateOldProperties(context);

    String topicProperty = context.getString(KafkaSourceConstants.TOPICS_REGEX);
    if (topicProperty != null && !topicProperty.isEmpty()) {
      // create subscriber that uses pattern-based subscription
      subscriber = new PatternSubscriber(topicProperty);
    } else if ((topicProperty = context.getString(KafkaSourceConstants.TOPICS)) != null &&
               !topicProperty.isEmpty()) {
      // create subscriber that uses topic list subscription
      subscriber = new TopicListSubscriber(topicProperty);
    } else if (subscriber == null) {
      throw new ConfigurationException("At least one Kafka topic must be specified.");
    }

    batchUpperLimit = context.getInteger(KafkaSourceConstants.BATCH_SIZE,
                                         KafkaSourceConstants.DEFAULT_BATCH_SIZE);
    maxBatchDurationMillis = context.getInteger(KafkaSourceConstants.BATCH_DURATION_MS,
                                                KafkaSourceConstants.DEFAULT_BATCH_DURATION);

    useAvroEventFormat = context.getBoolean(KafkaSourceConstants.AVRO_EVENT,
                                            KafkaSourceConstants.DEFAULT_AVRO_EVENT);

    if (log.isDebugEnabled()) {
      log.debug(KafkaSourceConstants.AVRO_EVENT + " set to: {}", useAvroEventFormat);
    }

    zookeeperConnect = context.getString(ZOOKEEPER_CONNECT_FLUME_KEY);
    migrateZookeeperOffsets = context.getBoolean(MIGRATE_ZOOKEEPER_OFFSETS,
        DEFAULT_MIGRATE_ZOOKEEPER_OFFSETS);

    bootstrapServers = context.getString(KafkaSourceConstants.BOOTSTRAP_SERVERS);
    if (bootstrapServers == null || bootstrapServers.isEmpty()) {
      if (zookeeperConnect == null || zookeeperConnect.isEmpty()) {
        throw new ConfigurationException("Bootstrap Servers must be specified");
      } else {
        // For backwards compatibility look up the bootstrap from zookeeper
        log.warn("{} is deprecated. Please use the parameter {}",
            KafkaSourceConstants.ZOOKEEPER_CONNECT_FLUME_KEY,
            KafkaSourceConstants.BOOTSTRAP_SERVERS);

        // Lookup configured security protocol, just in case its not default
        String securityProtocolStr =
            context.getSubProperties(KafkaSourceConstants.KAFKA_CONSUMER_PREFIX)
                .get(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG);
        if (securityProtocolStr == null || securityProtocolStr.isEmpty()) {
          securityProtocolStr = CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL;
        }
        bootstrapServers =
            lookupBootstrap(zookeeperConnect, SecurityProtocol.valueOf(securityProtocolStr));
      }
    }

    String groupIdProperty =
        context.getString(KAFKA_CONSUMER_PREFIX + ConsumerConfig.GROUP_ID_CONFIG);
    if (groupIdProperty != null && !groupIdProperty.isEmpty()) {
      groupId = groupIdProperty; // Use the new group id property
    }

    if (groupId == null || groupId.isEmpty()) {
      groupId = DEFAULT_GROUP_ID;
      log.info("Group ID was not specified. Using {} as the group id.", groupId);
    }

    setTopicHeader = context.getBoolean(KafkaSourceConstants.SET_TOPIC_HEADER,
                                        KafkaSourceConstants.DEFAULT_SET_TOPIC_HEADER);

    topicHeader = context.getString(KafkaSourceConstants.TOPIC_HEADER,
                                    KafkaSourceConstants.DEFAULT_TOPIC_HEADER);

    setConsumerProps(context);

    if (log.isDebugEnabled() && LogPrivacyUtil.allowLogPrintConfig()) {
      log.debug("Kafka consumer properties: {}", kafkaProps);
    }

    if (counter == null) {
      counter = new KafkaSourceCounter(getName());
    }
  }

}