@Test public void testSharedConfig() { Properties props = new Properties(); props.setProperty(KafkaWriterConfigurationKeys.KAFKA_PRODUCER_CONFIG_PREFIX + "key1", "value1"); props.setProperty(KafkaWriterConfigurationKeys.KAFKA_PRODUCER_CONFIG_PREFIX + "key2", "value2"); props.setProperty(ConfigurationKeys.SHARED_KAFKA_CONFIG_PREFIX + ".key1", "sharedValue1"); props.setProperty(ConfigurationKeys.SHARED_KAFKA_CONFIG_PREFIX + ".key3", "sharedValue3"); Properties producerProps = KafkaWriterHelper.getProducerProperties(props); Assert.assertEquals(producerProps.getProperty("key1"), "value1"); Assert.assertEquals(producerProps.getProperty("key2"), "value2"); Assert.assertEquals(producerProps.getProperty("key3"), "sharedValue3"); }
static Properties getProducerProperties(Properties props) { Config config = ConfigUtils.propertiesToConfig(props); Config producerConfig = ConfigUtils.getConfigOrEmpty(config, KAFKA_PRODUCER_CONFIG_PREFIX_NO_DOT).withFallback( ConfigUtils.getConfigOrEmpty(config, ConfigurationKeys.SHARED_KAFKA_CONFIG_PREFIX)); Properties producerProperties = ConfigUtils.configToProperties(producerConfig); setDefaultIfUnset(producerProperties, KEY_SERIALIZER_CONFIG, DEFAULT_KEY_SERIALIZER); setDefaultIfUnset(producerProperties, VALUE_SERIALIZER_CONFIG, DEFAULT_VALUE_SERIALIZER); setDefaultIfUnset(producerProperties, CLIENT_ID_CONFIG, CLIENT_ID_DEFAULT); setDefaultIfUnset(producerProperties, KAFKA_SCHEMA_REGISTRY_SWITCH_NAME, KAFKA_SCHEMA_REGISTRY_SWITCH_NAME_DEFAULT); return producerProperties; }
KafkaWriterHelper { static Properties getProducerProperties(Properties props) { Config config = ConfigUtils.propertiesToConfig(props); Config producerConfig = ConfigUtils.getConfigOrEmpty(config, KAFKA_PRODUCER_CONFIG_PREFIX_NO_DOT).withFallback( ConfigUtils.getConfigOrEmpty(config, ConfigurationKeys.SHARED_KAFKA_CONFIG_PREFIX)); Properties producerProperties = ConfigUtils.configToProperties(producerConfig); setDefaultIfUnset(producerProperties, KEY_SERIALIZER_CONFIG, DEFAULT_KEY_SERIALIZER); setDefaultIfUnset(producerProperties, VALUE_SERIALIZER_CONFIG, DEFAULT_VALUE_SERIALIZER); setDefaultIfUnset(producerProperties, CLIENT_ID_CONFIG, CLIENT_ID_DEFAULT); setDefaultIfUnset(producerProperties, KAFKA_SCHEMA_REGISTRY_SWITCH_NAME, KAFKA_SCHEMA_REGISTRY_SWITCH_NAME_DEFAULT); return producerProperties; } }
KafkaWriterHelper { static Properties getProducerProperties(Properties props) { Config config = ConfigUtils.propertiesToConfig(props); Config producerConfig = ConfigUtils.getConfigOrEmpty(config, KAFKA_PRODUCER_CONFIG_PREFIX_NO_DOT).withFallback( ConfigUtils.getConfigOrEmpty(config, ConfigurationKeys.SHARED_KAFKA_CONFIG_PREFIX)); Properties producerProperties = ConfigUtils.configToProperties(producerConfig); setDefaultIfUnset(producerProperties, KEY_SERIALIZER_CONFIG, DEFAULT_KEY_SERIALIZER); setDefaultIfUnset(producerProperties, VALUE_SERIALIZER_CONFIG, DEFAULT_VALUE_SERIALIZER); setDefaultIfUnset(producerProperties, CLIENT_ID_CONFIG, CLIENT_ID_DEFAULT); setDefaultIfUnset(producerProperties, KAFKA_SCHEMA_REGISTRY_SWITCH_NAME, KAFKA_SCHEMA_REGISTRY_SWITCH_NAME_DEFAULT); return producerProperties; }  }
KafkaWriterHelper { static Properties getProducerProperties(Properties props) { Config config = ConfigUtils.propertiesToConfig(props); Config producerConfig = ConfigUtils.getConfigOrEmpty(config, KAFKA_PRODUCER_CONFIG_PREFIX_NO_DOT).withFallback( ConfigUtils.getConfigOrEmpty(config, ConfigurationKeys.SHARED_KAFKA_CONFIG_PREFIX)); Properties producerProperties = ConfigUtils.configToProperties(producerConfig); setDefaultIfUnset(producerProperties, KEY_SERIALIZER_CONFIG, DEFAULT_KEY_SERIALIZER); setDefaultIfUnset(producerProperties, VALUE_SERIALIZER_CONFIG, DEFAULT_VALUE_SERIALIZER); setDefaultIfUnset(producerProperties, CLIENT_ID_CONFIG, CLIENT_ID_DEFAULT); setDefaultIfUnset(producerProperties, KAFKA_SCHEMA_REGISTRY_SWITCH_NAME, KAFKA_SCHEMA_REGISTRY_SWITCH_NAME_DEFAULT); return producerProperties; }  static Object getKafkaProducer(Properties props); static Pair<K,V> getKeyValuePair(V record, KafkaWriterCommonConfig commonConfig); }
KafkaWriterHelper { static Properties getProducerProperties(Properties props) { Config config = ConfigUtils.propertiesToConfig(props); Config producerConfig = ConfigUtils.getConfigOrEmpty(config, KAFKA_PRODUCER_CONFIG_PREFIX_NO_DOT).withFallback( ConfigUtils.getConfigOrEmpty(config, ConfigurationKeys.SHARED_KAFKA_CONFIG_PREFIX)); Properties producerProperties = ConfigUtils.configToProperties(producerConfig); setDefaultIfUnset(producerProperties, KEY_SERIALIZER_CONFIG, DEFAULT_KEY_SERIALIZER); setDefaultIfUnset(producerProperties, VALUE_SERIALIZER_CONFIG, DEFAULT_VALUE_SERIALIZER); setDefaultIfUnset(producerProperties, CLIENT_ID_CONFIG, CLIENT_ID_DEFAULT); setDefaultIfUnset(producerProperties, KAFKA_SCHEMA_REGISTRY_SWITCH_NAME, KAFKA_SCHEMA_REGISTRY_SWITCH_NAME_DEFAULT); return producerProperties; }  static Object getKafkaProducer(Properties props); static Pair<K,V> getKeyValuePair(V record, KafkaWriterCommonConfig commonConfig);  }