class ThriftSource extends SslContextAwareAbstractSource implements Configurable, EventDrivenSource {

  @Override
  public void configure(Context context) {
    configureSsl(context);
    logger.info("Configuring thrift source.");
    port = context.getInteger(CONFIG_PORT);
    Preconditions.checkNotNull(port, "Port must be specified for Thrift " +
        "Source.");
    bindAddress = context.getString(CONFIG_BIND);
    Preconditions.checkNotNull(bindAddress, "Bind address must be specified " +
        "for Thrift Source.");

    try {
      maxThreads = context.getInteger(CONFIG_THREADS, 0);
      maxThreads = (maxThreads <= 0) ? Integer.MAX_VALUE : maxThreads;
    } catch (NumberFormatException e) {
      logger.warn("Thrift source\'s \"threads\" property must specify an " +
                  "integer value: " + context.getString(CONFIG_THREADS));
    }

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

    protocol = context.getString(CONFIG_PROTOCOL);
    if (protocol == null) {
      // default is to use the compact protocol.
      protocol = COMPACT_PROTOCOL;
    }
    Preconditions.checkArgument(
        (protocol.equalsIgnoreCase(BINARY_PROTOCOL) ||
                protocol.equalsIgnoreCase(COMPACT_PROTOCOL)),
        "binary or compact are the only valid Thrift protocol types to " +
                "choose from.");

    principal = context.getString(AGENT_PRINCIPAL);
    String keytab = context.getString(AGENT_KEYTAB);
    enableKerberos = context.getBoolean(KERBEROS_KEY, false);
    this.flumeAuth = FlumeAuthenticationUtil.getAuthenticator(principal, keytab);
    if (enableKerberos) {
      if (!flumeAuth.isAuthenticated()) {
        throw new FlumeException("Authentication failed in Kerberos mode for " +
                "principal " + principal + " keytab " + keytab);
      }
      flumeAuth.startCredentialRefresher();
    }
  }

}