ProducerPropertiesFactory { Map<String, Object> getProducerProperties() { final HashMap<String, Object> producerProperties = configToProperties(kafkaConfig.getInternalProducerConfig().getConfig(KAFKA_CLIENTS_KEY)); addMetadata(producerProperties); addSecurityProtocol(producerProperties); addSpecificConfig(producerProperties); return Collections.unmodifiableMap(producerProperties); } private  ProducerPropertiesFactory(final Connection connection, final KafkaConfig kafkaConfig);   }
@Test public void addsBootstrapServersAndFlattensProperties() { final Map<String, Object> properties = underTest.getProducerProperties(); final List<String> servers = Arrays.asList(properties.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG).toString().split(",")); assertThat(servers).containsExactlyInAnyOrder(BOOTSTRAP_SERVERS); assertThat(properties).contains( new AbstractMap.SimpleEntry<>("connections.max.idle.ms", 543210), new AbstractMap.SimpleEntry<>("reconnect.backoff.ms", 500), new AbstractMap.SimpleEntry<>("reconnect.backoff.max.ms", 10000) ); }