### Question:
Adapters { public static <T> Consumer<T> toConsumer(SubscriberFactory<T> subscriberFactory) { return t -> Mono.just(t).subscribe(subscriberFactory.create()); } private  Adapters(); static Acknowledgeable<T> toLoggingAcknowledgeable(T t); static Subscriber<Acknowledgeable<T>> toAcknowledgeableSubscriber(Subscriber<T> subscriber); static Subscriber<T> toSendingSubscriber(Consumer<? super Flux<T>> sender); static Subscriber<T> toSubscriber(Consumer<T> consumer); static Consumer<T> toConsumer(SubscriberFactory<T> subscriberFactory); static Consumer<T> toSendingConsumerPerThread(Consumer<? super Flux<T>> sender); static Consumer<T> toSendingConsumer(Consumer<? super Flux<T>> sender); static Consumer<T> toSynchronousConsumer(Function<? super Flux<T>, ? extends Publisher<R>> sender); static Consumer<T> toSynchronousConsumer(Function<? super Flux<T>, ? extends Publisher<R>> sender, Duration timeout);  }### Answer:
@Test public void consumerDoesNotDeadlockOnFailureWithRetry() { FluxProcessor<String, String> processor = EmitterProcessor.create(1); handleWithErrorAndRetry(processor).subscribe(); Adapters.toConsumer(() -> processor).accept("Hello"); }