KafkaSSLUtil { public static void addGlobalSSLParameters(Properties kafkaProps) { if (isSSLEnabled(kafkaProps)) { addGlobalSSLParameter(kafkaProps, SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, SSLUtil.getGlobalKeystorePath()); addGlobalSSLParameter(kafkaProps, SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, SSLUtil.getGlobalKeystorePassword()); addGlobalSSLParameter(kafkaProps, SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, SSLUtil.getGlobalKeystoreType(null)); addGlobalSSLParameter(kafkaProps, SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, SSLUtil.getGlobalTruststorePath()); addGlobalSSLParameter(kafkaProps, SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, SSLUtil.getGlobalTruststorePassword()); addGlobalSSLParameter(kafkaProps, SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, SSLUtil.getGlobalTruststoreType(null)); } } private  KafkaSSLUtil(); static void addGlobalSSLParameters(Properties kafkaProps);  }
@Test public void testSecurityProtocol_PLAINTEXT() { Properties kafkaProps = new Properties(); kafkaProps.put( CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.PLAINTEXT.name); KafkaSSLUtil.addGlobalSSLParameters(kafkaProps); assertNoSSLParameters(kafkaProps); }
@Test public void testSecurityProtocol_SASL_PLAINTEXT() { Properties kafkaProps = new Properties(); kafkaProps.put( CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SASL_PLAINTEXT.name); KafkaSSLUtil.addGlobalSSLParameters(kafkaProps); assertNoSSLParameters(kafkaProps); }
@Test public void testSecurityProtocol_SSL() { Properties kafkaProps = new Properties(); kafkaProps.put( CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name); KafkaSSLUtil.addGlobalSSLParameters(kafkaProps); assertGlobalSSLParameters(kafkaProps); }
@Test public void testSecurityProtocol_SASL_SSL() { Properties kafkaProps = new Properties(); kafkaProps.put( CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SASL_SSL.name); KafkaSSLUtil.addGlobalSSLParameters(kafkaProps); assertGlobalSSLParameters(kafkaProps); }
@Test public void testComponentParametersNotOverridden() { Properties kafkaProps = new Properties(); kafkaProps.put( CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name); kafkaProps.put( SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, "component-keystore-path"); kafkaProps.put( SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, "component-keystore-password"); kafkaProps.put( SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, "component-keystore-type"); kafkaProps.put( SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "component-truststore-path"); kafkaProps.put( SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "component-truststore-password"); kafkaProps.put( SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, "component-truststore-type"); KafkaSSLUtil.addGlobalSSLParameters(kafkaProps); assertComponentSSLParameters(kafkaProps); }
@Test public void testEmptyGlobalParametersNotAdded() { Properties kafkaProps = new Properties(); kafkaProps.put( CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name); clearSystemProperties(); KafkaSSLUtil.addGlobalSSLParameters(kafkaProps); assertNoSSLParameters(kafkaProps); }