@Nonnull
    public <K, V> Consumer<K, V> newConsumer() {
        return newConsumer(EMPTY_PROPERTIES);
    }
@Test
    public void should_create_new_consumer_for_each_call() {
        kafkaDataConnection = createNonSharedKafkaDataConnection();

        try (Consumer<Object, Object> c1 = kafkaDataConnection.newConsumer();
             Consumer<Object, Object> c2 = kafkaDataConnection.newConsumer()) {
            assertThat(c1).isNotSameAs(c2);
        }
    }