### Question:
Adapters { public static <T> Subscriber<Acknowledgeable<T>> toAcknowledgeableSubscriber(Subscriber<T> subscriber) { return toSendingSubscriber(flux -> flux.concatMap(Adapters::toTerminalAcknowledgement).subscribe(subscriber)); } 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 adaptedAcknowledgeableSubscribersAcknowledgeOnDownstreamError() { TestAcknowledgeable acknowledgeable = new TestAcknowledgeable("DATA"); FluxProcessor<String, String> downstream = EmitterProcessor.create(1); downstream.map(string -> { throw new IllegalArgumentException(); }).retry().subscribe(); FluxProcessor<Acknowledgeable<String>, Acknowledgeable<String>> upstream = EmitterProcessor.create(1); upstream.subscribe(Adapters.toAcknowledgeableSubscriber(downstream)); upstream.sink().next(acknowledgeable); assertTrue(acknowledgeable.isAcknowledged()); }