KafkaClientConfigUtil { public Map<String, Object> applyCommonSettings(final ClusterConfig clusterConfig, final String consumerId) { return applyCommonSettings(clusterConfig, consumerId, new HashMap<>()); } KafkaClientConfigUtil(final String keyStoreRootPath, final String consumerIdPrefix); Map<String, Object> applyCommonSettings(final ClusterConfig clusterConfig, final String consumerId); static List<KafkaSettings> getAllKafkaConsumerProperties();  }
@Test public void testApplyCommonSettings_noSsl_noSasl() { final ClusterConfig clusterConfig = ClusterConfig.newBuilder() .withBrokerHosts(expectedBrokerHosts) .withUseSsl(false) .withUseSasl(false) .build(); final Map<String, Object> config = util.applyCommonSettings(clusterConfig, consumerId); validateDefaultKeys(config); validateNoSsl(config); validateNoSasl(config); validateNoKey(config, CommonClientConfigs.SECURITY_PROTOCOL_CONFIG); }
@Test public void testApplyCommonSettings_noSsl_noSasl_withClientProperties() { final Map<String, String> customProperties = new HashMap<>(); customProperties.put("key3", "value3"); customProperties.put("key4", "value4"); final ClusterConfig clusterConfig = ClusterConfig.newBuilder() .withBrokerHosts(expectedBrokerHosts) .withUseSsl(false) .withUseSasl(false) .withClusterClientConfig("key1", "value1") .withClusterClientConfig("key2", "value2") .withClusterClientConfig(customProperties) .build(); final Map<String, Object> config = util.applyCommonSettings(clusterConfig, consumerId); validateDefaultKeys(config); validateNoSsl(config); validateNoSasl(config); validateNoKey(config, CommonClientConfigs.SECURITY_PROTOCOL_CONFIG); validateKey(config, "key1", "value1"); validateKey(config, "key2", "value2"); validateKey(config, "key3", "value3"); validateKey(config, "key4", "value4"); }
@Test public void testApplyCommonSettings_withSsl_noSasl_withTrustStore() { final ClusterConfig clusterConfig = ClusterConfig.newBuilder() .withBrokerHosts(expectedBrokerHosts) .withUseSsl(true) .withKeyStoreFile(expectedKeyStoreFile) .withKeyStorePassword(expectedKeyStorePassword) .withTrustStoreFile(expectedTrustStoreFile) .withTrustStorePassword(expectedTrustStorePassword) .withUseSasl(false) .build(); final Map<String, Object> config = util.applyCommonSettings(clusterConfig, consumerId); validateDefaultKeys(config); validateSsl(config, "SSL", true, true); validateNoSasl(config); }
@Test public void testApplyCommonSettings_withSsl_noSasl_withNoTrustStore() { final ClusterConfig clusterConfig = ClusterConfig.newBuilder() .withBrokerHosts(expectedBrokerHosts) .withUseSsl(true) .withKeyStoreFile(expectedKeyStoreFile) .withKeyStorePassword(expectedKeyStorePassword) .withTrustStoreFile(null) .withTrustStorePassword(null) .withUseSasl(false) .build(); final Map<String, Object> config = util.applyCommonSettings(clusterConfig, consumerId); validateDefaultKeys(config); validateSsl(config, "SSL", true, false); validateNoSasl(config); }
@Test public void testApplyCommonSettings_noSsl_withSasl_customMechanism() { final ClusterConfig clusterConfig = ClusterConfig.newBuilder() .withBrokerHosts(expectedBrokerHosts) .withUseSsl(false) .withUseSasl(true) .withSaslMechanism(expectedSaslCustomMechanism) .withSaslJaas(expectedSaslJaas) .build(); final Map<String, Object> config = util.applyCommonSettings(clusterConfig, consumerId); validateDefaultKeys(config); validateNoSsl(config); validateSaslCustomMechanism(config); }
@Test public void testApplyCommonSettings_noSsl_withSasl_plainMechanism() { final ClusterConfig clusterConfig = ClusterConfig.newBuilder() .withBrokerHosts(expectedBrokerHosts) .withUseSsl(false) .withUseSasl(true) .withSaslMechanism(expectedSaslPlainMechanism) .withSaslPlaintextUsername(expectedSaslUsername) .withSaslPlaintextPassword(expectedSaslPassword) .withSaslJaas(expectedSaslJaas) .build(); final Map<String, Object> config = util.applyCommonSettings(clusterConfig, consumerId); validateDefaultKeys(config); validateNoSsl(config); validateSaslPlainMechanism(config, "SASL_PLAINTEXT"); }
@Test public void testApplyCommonSettings_withSsl_withSasl_withTrustStore() { final ClusterConfig clusterConfig = ClusterConfig.newBuilder() .withBrokerHosts(expectedBrokerHosts) .withUseSsl(true) .withKeyStoreFile(expectedKeyStoreFile) .withKeyStorePassword(expectedKeyStorePassword) .withTrustStoreFile(expectedTrustStoreFile) .withTrustStorePassword(expectedTrustStorePassword) .withUseSasl(true) .withSaslMechanism(expectedSaslPlainMechanism) .withSaslPlaintextUsername(expectedSaslUsername) .withSaslPlaintextPassword(expectedSaslPassword) .withSaslJaas(expectedSaslJaas) .build(); final Map<String, Object> config = util.applyCommonSettings(clusterConfig, consumerId); validateDefaultKeys(config); validateSsl(config, "SASL_SSL", false, true); validateSaslPlainMechanism(config, "SASL_SSL"); }
@Test public void testApplyCommonSettings_withSsl_withSasl_withNoTrustStore() { final ClusterConfig clusterConfig = ClusterConfig.newBuilder() .withBrokerHosts(expectedBrokerHosts) .withUseSsl(true) .withKeyStoreFile(expectedKeyStoreFile) .withKeyStorePassword(expectedKeyStorePassword) .withTrustStoreFile(null) .withTrustStorePassword(null) .withUseSasl(true) .withSaslMechanism(expectedSaslPlainMechanism) .withSaslPlaintextUsername(expectedSaslUsername) .withSaslPlaintextPassword(expectedSaslPassword) .withSaslJaas(expectedSaslJaas) .build(); final Map<String, Object> config = util.applyCommonSettings(clusterConfig, consumerId); validateDefaultKeys(config); validateSsl(config, "SASL_SSL", false, false); validateSaslPlainMechanism(config, "SASL_SSL"); }