@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); }
@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(); } String strBatchSize = properties.getProperty( RpcClientConfigurationConstants.CONFIG_BATCH_SIZE); logger.debug("Batch size string = " + strBatchSize); batchSize = RpcClientConfigurationConstants.DEFAULT_BATCH_SIZE; if (strBatchSize != null && !strBatchSize.isEmpty()) { try { int parsedBatch = Integer.parseInt(strBatchSize); if (parsedBatch < 1) { logger.warn("Invalid value for batchSize: {}; Using default value.", parsedBatch); } else { batchSize = parsedBatch; } } catch (NumberFormatException e) { logger.warn("Batchsize is not valid for RpcClient: " + strBatchSize + ". Default value assigned.", e); } } 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); } } } enableSsl = Boolean.parseBoolean(properties.getProperty( RpcClientConfigurationConstants.CONFIG_SSL)); trustAllCerts = Boolean.parseBoolean(properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUST_ALL_CERTS)); truststore = properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUSTSTORE); truststorePassword = properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUSTSTORE_PASSWORD); truststoreType = properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUSTSTORE_TYPE, "JKS"); this.connect(); }
NettyAvroRpcClient extends AbstractRpcClient implements RpcClient { @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(); } String strBatchSize = properties.getProperty( RpcClientConfigurationConstants.CONFIG_BATCH_SIZE); logger.debug("Batch size string = " + strBatchSize); batchSize = RpcClientConfigurationConstants.DEFAULT_BATCH_SIZE; if (strBatchSize != null && !strBatchSize.isEmpty()) { try { int parsedBatch = Integer.parseInt(strBatchSize); if (parsedBatch < 1) { logger.warn("Invalid value for batchSize: {}; Using default value.", parsedBatch); } else { batchSize = parsedBatch; } } catch (NumberFormatException e) { logger.warn("Batchsize is not valid for RpcClient: " + strBatchSize + ". Default value assigned.", e); } } 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); } } } enableSsl = Boolean.parseBoolean(properties.getProperty( RpcClientConfigurationConstants.CONFIG_SSL)); trustAllCerts = Boolean.parseBoolean(properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUST_ALL_CERTS)); truststore = properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUSTSTORE); truststorePassword = properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUSTSTORE_PASSWORD); truststoreType = properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUSTSTORE_TYPE, "JKS"); this.connect(); } }
NettyAvroRpcClient extends AbstractRpcClient implements RpcClient { @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(); } String strBatchSize = properties.getProperty( RpcClientConfigurationConstants.CONFIG_BATCH_SIZE); logger.debug("Batch size string = " + strBatchSize); batchSize = RpcClientConfigurationConstants.DEFAULT_BATCH_SIZE; if (strBatchSize != null && !strBatchSize.isEmpty()) { try { int parsedBatch = Integer.parseInt(strBatchSize); if (parsedBatch < 1) { logger.warn("Invalid value for batchSize: {}; Using default value.", parsedBatch); } else { batchSize = parsedBatch; } } catch (NumberFormatException e) { logger.warn("Batchsize is not valid for RpcClient: " + strBatchSize + ". Default value assigned.", e); } } 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); } } } enableSsl = Boolean.parseBoolean(properties.getProperty( RpcClientConfigurationConstants.CONFIG_SSL)); trustAllCerts = Boolean.parseBoolean(properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUST_ALL_CERTS)); truststore = properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUSTSTORE); truststorePassword = properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUSTSTORE_PASSWORD); truststoreType = properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUSTSTORE_TYPE, "JKS"); this.connect(); } protected  NettyAvroRpcClient(); }
NettyAvroRpcClient extends AbstractRpcClient implements RpcClient { @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(); } String strBatchSize = properties.getProperty( RpcClientConfigurationConstants.CONFIG_BATCH_SIZE); logger.debug("Batch size string = " + strBatchSize); batchSize = RpcClientConfigurationConstants.DEFAULT_BATCH_SIZE; if (strBatchSize != null && !strBatchSize.isEmpty()) { try { int parsedBatch = Integer.parseInt(strBatchSize); if (parsedBatch < 1) { logger.warn("Invalid value for batchSize: {}; Using default value.", parsedBatch); } else { batchSize = parsedBatch; } } catch (NumberFormatException e) { logger.warn("Batchsize is not valid for RpcClient: " + strBatchSize + ". Default value assigned.", e); } } 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); } } } enableSsl = Boolean.parseBoolean(properties.getProperty( RpcClientConfigurationConstants.CONFIG_SSL)); trustAllCerts = Boolean.parseBoolean(properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUST_ALL_CERTS)); truststore = properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUSTSTORE); truststorePassword = properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUSTSTORE_PASSWORD); truststoreType = properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUSTSTORE_TYPE, "JKS"); 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); }
NettyAvroRpcClient extends AbstractRpcClient implements RpcClient { @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(); } String strBatchSize = properties.getProperty( RpcClientConfigurationConstants.CONFIG_BATCH_SIZE); logger.debug("Batch size string = " + strBatchSize); batchSize = RpcClientConfigurationConstants.DEFAULT_BATCH_SIZE; if (strBatchSize != null && !strBatchSize.isEmpty()) { try { int parsedBatch = Integer.parseInt(strBatchSize); if (parsedBatch < 1) { logger.warn("Invalid value for batchSize: {}; Using default value.", parsedBatch); } else { batchSize = parsedBatch; } } catch (NumberFormatException e) { logger.warn("Batchsize is not valid for RpcClient: " + strBatchSize + ". Default value assigned.", e); } } 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); } } } enableSsl = Boolean.parseBoolean(properties.getProperty( RpcClientConfigurationConstants.CONFIG_SSL)); trustAllCerts = Boolean.parseBoolean(properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUST_ALL_CERTS)); truststore = properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUSTSTORE); truststorePassword = properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUSTSTORE_PASSWORD); truststoreType = properties.getProperty( RpcClientConfigurationConstants.CONFIG_TRUSTSTORE_TYPE, "JKS"); 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);  }