### Question:
SnowdropKafkaProducer implements KafkaProducer<K, V> { @Override @SuppressWarnings("unchecked") public <T> Mono<T> doOnVertxProducer(Function<io.vertx.kafka.client.producer.KafkaProducer<K, V>, T> function) { Objects.requireNonNull(function, "Function cannot be null"); return Mono.create(sink -> { try { T result = function.apply((io.vertx.kafka.client.producer.KafkaProducer<K, V>) delegate.getDelegate()); sink.success(result); } 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 public void shouldDoOnVertxProducer() { given(mockMutinyProducer.getDelegate()) .willReturn(mockVertxProducer); AtomicReference<io.vertx.kafka.client.producer.KafkaProducer<Integer, String>> vertxConsumer = new AtomicReference<>(); Function<io.vertx.kafka.client.producer.KafkaProducer<Integer, String>, Boolean> function = vp -> { vertxConsumer.set(vp); return true; }; StepVerifier.create(producer.doOnVertxProducer(function)) .expectNext(true) .verifyComplete(); assertThat(vertxConsumer.get()).isEqualTo(mockVertxProducer); }