ConsumerRecordRelay implements Runnable { void stop() { logger.info("Stopping Kafka message relay"); stopped = true; } ConsumerRecordRelay(Consumer<K, V> consumer, BlockingQueueConsumer<K, V> blockingQueueConsumer); @Override void run(); void setOffset(ConsumerRecord<K, V> record); void removePartitionFromOffset(TopicPartition topicPartition);  }
@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(); }