@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);
    // check flattening of client properties in kafka.producer.internal.kafka-clients
    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)
    );
}