Processor implements Runnable { public void queue(ConsumerRecord<K, V> record) throws InterruptedException { queue.put(record); } Processor(TopicPartition topicPartition, ConsumerRecordRelay<K, V> relay, java.util.function.Consumer<V> action,
            int queueSize); @Override void run(); void stop(); void queue(ConsumerRecord<K, V> record); boolean isStopped();  }
@Test public void stoppedSetOnException() throws Exception { final Processor<Integer, String> processor = new Processor<>(new TopicPartition("Hello", 1), relay, message -> { throw new RuntimeException("Foobar! Something went wrong"); }, 42); new Thread(processor).start(); processor.queue(record); await().until(processor::isStopped, is(equalTo(true))); }