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); 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); 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); } } 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(); @Override void close(); @Override String toString(); @Override void append(Event event); @Override void appendBatch(List<Event> events); @Override boolean isActive(); @Override synchronized void configure(Properties properties);  }
@Test(expected = FlumeException.class) public void testUnableToConnect() throws FlumeException { @SuppressWarnings("unused") NettyAvroRpcClient client = new NettyAvroRpcClient(); Properties props = new Properties(); props.setProperty(RpcClientConfigurationConstants.CONFIG_HOSTS, "localhost"); props.setProperty(RpcClientConfigurationConstants.CONFIG_HOSTS_PREFIX + "localhost", localhost + ":" + 1); client.configure(props); }