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

        assertThatThrownBy(() -> kafkaDataConnection.newConsumer())
                .isInstanceOf(IllegalArgumentException.class)
                .hasMessage("KafkaConsumer is not thread-safe and can't be used"
                        + " with shared DataConnection 'kafka-data-connection'");

        assertThatThrownBy(() -> kafkaDataConnection.newConsumer(new Properties()))
                .isInstanceOf(IllegalArgumentException.class)
                .hasMessage("KafkaConsumer is not thread-safe and can't be used"
                        + " with shared DataConnection 'kafka-data-connection'");
    }