    @Test
    void errorInMessageHandlingKillsSubscription(KafkaHelper kafkaHelper) throws Exception {

        String topic = "topic4SubsErrorTest";
        ensureTopicsExists(topic);

        // send two good and one bad message
        KafkaProducer<String, String> stringProducer = kafkaHelper.createStringProducer();
        stringProducer.send(new ProducerRecord<>(topic, "1")).get();
        logger.error("Sent 1");
        stringProducer.send(new ProducerRecord<>(topic, "Two")).get();
        logger.error("Sent Two");
        stringProducer.send(new ProducerRecord<>(topic, "3")).get();
        logger.error("Sent 3");

        AtomicInteger counter = new AtomicInteger();
        List<Integer> messagesFromSut = new ArrayList<>();
        sut.messages(topic).subscribe(
                x -> {
                    logger.error("Received " + x);
                    try {
                        messagesFromSut.add(Integer.parseInt(x));
                    } finally {
                        counter.incrementAndGet();
                    }
                }
        );

        List<String> messagesFromBroker = kafkaHelper.consumeStrings(topic, 3).get(20, SECONDS);
        assertThat(messagesFromBroker, contains("1", "Two", "3"));
        // sleep a few seconds more to make sure no more messages wil be consumed
        SECONDS.sleep(10);
        assertThat(counter.get(),is(2));
        assertThat(messagesFromSut, containsInAnyOrder(1));
    }