class NettyAvroRpcClient extends SSLContextAwareAbstractRpcClient {

  @Override
  public synchronized void configure(Properties properties)
      throws FlumeException {
    stateLock.lock();
    try {
      if (connState == ConnState.READY || connState == ConnState.DEAD) {
        throw new FlumeException("This client was already configured, " +
            "cannot reconfigure.");
      }
    } finally {
      stateLock.unlock();
    }

    batchSize = parseBatchSize(properties);

    // host and port
    String hostNames = properties.getProperty(
        RpcClientConfigurationConstants.CONFIG_HOSTS);
    String[] hosts = null;
    if (hostNames != null && !hostNames.isEmpty()) {
      hosts = hostNames.split("\\s+");
    } else {
      throw new FlumeException("Hosts list is invalid: " + hostNames);
    }

    if (hosts.length > 1) {
      logger.warn("More than one hosts are specified for the default client. "
          + "Only the first host will be used and others ignored. Specified: "
          + hostNames + "; to be used: " + hosts[0]);
    }

    String host = properties.getProperty(
        RpcClientConfigurationConstants.CONFIG_HOSTS_PREFIX + hosts[0]);
    if (host == null || host.isEmpty()) {
      throw new FlumeException("Host not found: " + hosts[0]);
    }
    String[] hostAndPort = host.split(":");
    if (hostAndPort.length != 2) {
      throw new FlumeException("Invalid hostname: " + hosts[0]);
    }
    Integer port = null;
    try {
      port = Integer.parseInt(hostAndPort[1]);
    } catch (NumberFormatException e) {
      throw new FlumeException("Invalid Port: " + hostAndPort[1], e);
    }
    this.address = new InetSocketAddress(hostAndPort[0], port);

    // connect timeout
    connectTimeout =
        RpcClientConfigurationConstants.DEFAULT_CONNECT_TIMEOUT_MILLIS;
    String strConnTimeout = properties.getProperty(
        RpcClientConfigurationConstants.CONFIG_CONNECT_TIMEOUT);
    if (strConnTimeout != null && strConnTimeout.trim().length() > 0) {
      try {
        connectTimeout = Long.parseLong(strConnTimeout);
        if (connectTimeout < 1000) {
          logger.warn("Connection timeout specified less than 1s. " +
              "Using default value instead.");
          connectTimeout =
              RpcClientConfigurationConstants.DEFAULT_CONNECT_TIMEOUT_MILLIS;
        }
      } catch (NumberFormatException ex) {
        logger.error("Invalid connect timeout specified: " + strConnTimeout);
      }
    }

    // request timeout
    requestTimeout =
        RpcClientConfigurationConstants.DEFAULT_REQUEST_TIMEOUT_MILLIS;
    String strReqTimeout = properties.getProperty(
        RpcClientConfigurationConstants.CONFIG_REQUEST_TIMEOUT);
    if  (strReqTimeout != null && strReqTimeout.trim().length() > 0) {
      try {
        requestTimeout = Long.parseLong(strReqTimeout);
        if (requestTimeout < 1000) {
          logger.warn("Request timeout specified less than 1s. " +
              "Using default value instead.");
          requestTimeout =
              RpcClientConfigurationConstants.DEFAULT_REQUEST_TIMEOUT_MILLIS;
        }
      } catch (NumberFormatException ex) {
        logger.error("Invalid request timeout specified: " + strReqTimeout);
      }
    }

    String enableCompressionStr =
        properties.getProperty(RpcClientConfigurationConstants.CONFIG_COMPRESSION_TYPE);
    if (enableCompressionStr != null && enableCompressionStr.equalsIgnoreCase("deflate")) {
      this.enableDeflateCompression = true;
      String compressionLvlStr =
          properties.getProperty(RpcClientConfigurationConstants.CONFIG_COMPRESSION_LEVEL);
      compressionLevel = RpcClientConfigurationConstants.DEFAULT_COMPRESSION_LEVEL;
      if (compressionLvlStr != null) {
        try {
          compressionLevel = Integer.parseInt(compressionLvlStr);
        } catch (NumberFormatException ex) {
          logger.error("Invalid compression level: " + compressionLvlStr);
        }
      }
    }

    configureSSL(properties);

    String maxIoWorkersStr = properties.getProperty(RpcClientConfigurationConstants.MAX_IO_WORKERS);
    if (!StringUtils.isEmpty(maxIoWorkersStr)) {
      try {
        maxIoWorkers = Integer.parseInt(maxIoWorkersStr);
      } catch (NumberFormatException ex) {
        logger.warn("Invalid maxIOWorkers:" + maxIoWorkersStr + " Using " +
            "default maxIOWorkers.");
        maxIoWorkers = -1;
      }
    }

    if (maxIoWorkers < 1) {
      logger.info("Using default maxIOWorkers");
      maxIoWorkers = -1;
    }

    this.connect();
  }

  protected  NettyAvroRpcClient();

}