class AvroSource extends SslContextAwareAbstractSource implements EventDrivenSource,
    Configurable, AvroSourceProtocol {

  @Override
  public void configure(Context context) {
    configureSsl(context);
    Configurables.ensureRequiredNonNull(context, PORT_KEY, BIND_KEY);

    port = context.getInteger(PORT_KEY);
    bindAddress = context.getString(BIND_KEY);
    compressionType = context.getString(COMPRESSION_TYPE, "none");

    try {
      maxThreads = context.getInteger(THREADS, 0);
    } catch (NumberFormatException e) {
      logger.warn("AVRO source\'s \"threads\" property must specify an integer value.",
              context.getString(THREADS));
    }

    enableIpFilter = context.getBoolean(IP_FILTER_KEY, false);
    if (enableIpFilter) {
      patternRuleConfigDefinition = context.getString(IP_FILTER_RULES_KEY);
      if (patternRuleConfigDefinition == null ||
          patternRuleConfigDefinition.trim().isEmpty()) {
        throw new FlumeException(
          "ipFilter is configured with true but ipFilterRules is not defined:" +
            " ");
      }
      String[] patternRuleDefinitions = patternRuleConfigDefinition.split(
        ",");
      rules = new ArrayList<IpFilterRule>(patternRuleDefinitions.length);
      for (String patternRuleDefinition : patternRuleDefinitions) {
        rules.add(generateRule(patternRuleDefinition));
      }
    }

    if (sourceCounter == null) {
      sourceCounter = new SourceCounter(getName());
    }
  }

}