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()); } }  @Override void configure(Context context); @Override void start(); @Override void stop(); @Override String toString(); @Override Status append(AvroFlumeEvent avroEvent); @Override Status appendBatch(List<AvroFlumeEvent> events);  }
@Test public void testSslRequestWithComponentKeystore() throws InterruptedException, IOException { Context context = new Context(); context.put("port", String.valueOf(selectedPort = getFreePort())); context.put("bind", "0.0.0.0"); context.put("ssl", "true"); context.put("keystore", "src/test/resources/server.p12"); context.put("keystore-password", "password"); context.put("keystore-type", "PKCS12"); Configurables.configure(source, context); doSslRequest(); }
@Test public void testSslRequestWithGlobalKeystore() throws InterruptedException, IOException { System.setProperty("javax.net.ssl.keyStore", "src/test/resources/server.p12"); System.setProperty("javax.net.ssl.keyStorePassword", "password"); System.setProperty("javax.net.ssl.keyStoreType", "PKCS12"); Context context = new Context(); context.put("port", String.valueOf(selectedPort = getFreePort())); context.put("bind", "0.0.0.0"); context.put("ssl", "true"); Configurables.configure(source, context); doSslRequest(); System.clearProperty("javax.net.ssl.keyStore"); System.clearProperty("javax.net.ssl.keyStorePassword"); }