class ConsumerRecordRelay implements Runnable {

    void stop() {
        logger.info("Stopping Kafka message relay");
        stopped = true;
    }

}

class ConsumerRecordRelayTest {

    @Mock
    private Consumer<Integer, String> consumer;
    @Mock
    private BlockingQueueConsumer<Integer, String> blockingQueueConsumer;
    private ConsumerRecord<Integer, String> record;

    @Test
    public void relayWithException() throws Exception {

        when(consumer.poll(anyLong())).thenThrow(RuntimeException.class);

        ConsumerRecordRelay<Integer, String> relay = new ConsumerRecordRelay<>(consumer, blockingQueueConsumer);
        new Thread(relay).start();
        verify(blockingQueueConsumer, never()).relay(record);
        verify(consumer, timeout(1000).only()).poll(anyLong());
        relay.stop();
    }

}