### Question:
SnowdropKafkaProducer implements KafkaProducer<K, V> { @Override public Mono<Void> flush() { return Mono.create(sink -> { try { delegate.flush(v -> sink.success()); } catch (Throwable t) { sink.error(t); } }); } SnowdropKafkaProducer(io.vertx.mutiny.kafka.client.producer.KafkaProducer<K, V> delegate); @Override Mono<RecordMetadata> send(ProducerRecord<K, V> record); @Override Flux<PartitionInfo> partitionsFor(String topic); @Override Mono<Void> flush(); @Override Mono<Void> close(); @Override Mono<Void> close(long timeout); @Override @SuppressWarnings("unchecked") // Mutiny API returns KafkaProducer without generics Mono<T> doOnVertxProducer(Function<io.vertx.kafka.client.producer.KafkaProducer<K, V>, T> function); @Override KafkaProducer<K, V> exceptionHandler(Consumer<Throwable> handler); @Override KafkaProducer<K, V> drainHandler(Consumer<Void> handler); @Override KafkaProducer<K, V> setWriteQueueMaxSize(int maxSize); @Override boolean writeQueueFull(); @Override Mono<Void> write(ProducerRecord<K, V> record); @Override Mono<Void> end(); @Override WriteStream vertxWriteStream();  }### Answer:
@Test @SuppressWarnings("unchecked") public void shouldFlush() { given(mockMutinyProducer.flush(any())) .will(a -> { ((Consumer<Void>) a.getArgument(0)).accept(null); return mockMutinyProducer; }); StepVerifier.create(producer.flush()) .verifyComplete(); }
@Test public void shouldHandleFlushFailure() { given(mockMutinyProducer.flush(any())) .willThrow(new RuntimeException("test")); StepVerifier.create(producer.flush()) .verifyErrorMessage("test"); }