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) { 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(); } }  @Override void configure(Context context); @Override void start(); @Override void stop(); static final Logger logger; static final String CONFIG_THREADS; static final String CONFIG_BIND; static final String CONFIG_PORT; static final String CONFIG_PROTOCOL; static final String BINARY_PROTOCOL; static final String COMPACT_PROTOCOL; }
@Test public void testAppendSSLWithComponentKeystore() throws Exception { Context context = new Context(); channel.configure(context); configureSource(); context.put(ThriftSource.CONFIG_BIND, "0.0.0.0"); context.put(ThriftSource.CONFIG_PORT, String.valueOf(port)); context.put("ssl", "true"); context.put("keystore", "src/test/resources/keystorefile.jks"); context.put("keystore-password", "password"); context.put("keystore-type", "JKS"); Configurables.configure(source, context); doAppendSSL(); }
@Test public void testAppendSSLWithGlobalKeystore() throws Exception { System.setProperty("javax.net.ssl.keyStore", "src/test/resources/keystorefile.jks"); System.setProperty("javax.net.ssl.keyStorePassword", "password"); System.setProperty("javax.net.ssl.keyStoreType", "JKS"); Context context = new Context(); channel.configure(context); configureSource(); context.put(ThriftSource.CONFIG_BIND, "0.0.0.0"); context.put(ThriftSource.CONFIG_PORT, String.valueOf(port)); context.put("ssl", "true"); Configurables.configure(source, context); doAppendSSL(); System.clearProperty("javax.net.ssl.keyStore"); System.clearProperty("javax.net.ssl.keyStorePassword"); System.clearProperty("javax.net.ssl.keyStoreType"); }