KafkaCommon { public static void addCustomParams(Map<String, Object> params, Config config) { for (Map.Entry<String, ConfigValue> entry : config.entrySet()) { String propertyName = entry.getKey(); if (propertyName.startsWith(PARAMETER_CONFIG_PREFIX)) { String paramName = propertyName.substring(PARAMETER_CONFIG_PREFIX.length()); String paramValue = config.getString(propertyName); if (LOG.isDebugEnabled()) { LOG.debug("Adding Kafka property: {} = \"{}\"", paramName, paramValue); } params.put(paramName, paramValue); } } }  static void addCustomParams(Map<String, Object> params, Config config); static final String PARAMETER_CONFIG_PREFIX; }
@Test public void testCustomParams() throws Exception { Map<String, Object> kafkaParams = Maps.newHashMap(); KafkaCommon.addCustomParams(kafkaParams, appConfig); assertEquals("SSL", kafkaParams.get("security.protocol")); assertEquals("/path/to/truststore.jks", kafkaParams.get("ssl.truststore.location")); assertEquals("changeme", kafkaParams.get("ssl.truststore.password")); }