Demonstrates that element hangs in time buffered WorkQueueProcessor if it is submitted at the same time when other elements are consumed. Completion with exception means problem reproduction. Java 8 and Reactor 2.5.0.M4 are used. Async Log4j2 with SLF4J is used for logging. Logging can be comment out to simplify setup.

HangingElement.java

package org.ivan.experiments.reactor;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.WorkQueueProcessor;
import reactor.core.scheduler.Schedulers;
import reactor.core.scheduler.TimedScheduler;
import reactor.core.subscriber.SubmissionEmitter;

import java.io.File;
import java.time.Duration;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;

import static reactor.core.util.WaitStrategy.liteBlocking;

public class HangingElement {

    static {
        System.setProperty("log4j.configurationFile", new File("log4j.xml").getAbsolutePath());
        System.setProperty("Log4jContextSelector", "org.apache.logging.log4j.core.async.AsyncLoggerContextSelector");
        new File("debug.log").delete();
    }

    public static final Logger logger = LoggerFactory.getLogger(HangingElement.class);

    public static final String e = "Element";
    public static final String s = "Synchronizer";

    public static void main(String[] args) throws Exception {
        highRate();
    }

    public static void highRate() throws Exception {
        WorkQueueProcessor<String> queueProcessor = WorkQueueProcessor.share("Processor", 256, liteBlocking());
        TimedScheduler timer = Schedulers.newTimer("Timer", 1);
        queueProcessor
                .buffer(32, Duration.ofMillis(2), timer)
                .subscribe(new Subscriber<List<String>>() {
                    int counter;
                    @Override
                    public void onSubscribe(Subscription s) {
                        s.request(Long.MAX_VALUE);
                    }

                    @Override
                    public void onNext(List<String> strings) {
                        int size = strings.size();
                        logger.debug("Consumed: " + size);
                        counter += size;
                        if (strings.contains(s)) {
                            synchronized (s) {
                                logger.debug("Synchronizer!");
                                s.notifyAll();
                            }
                        }
                    }

                    @Override
                    public void onError(Throwable t) {
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("Consumed in total: " + counter);
                    }
                });
        SubmissionEmitter<String> emitter = queueProcessor.connectEmitter();

        try {
            submitInCurrentThread(emitter);
        } finally {
            emitter.finish();
            timer.shutdown();
        }
        TimeUnit.SECONDS.sleep(1);
    }

    public static void submitInCurrentThread(SubmissionEmitter<String> emitter) {
        Random rand = new Random();
        for (int i = 0; i < 100_000; i++) {
            long re = emitter.submit(e);
            logger.debug("Submit element result " + re);
            LockSupport.parkNanos(2_000_000 + rand.nextInt(200_000) - 100_000);
            synchronized (s) {
                long rd = emitter.submit(s);
                logger.debug("Submit drain result " + rd);
                timeoutWait(s);
            }
        }
    }

    private static void timeoutWait(Object o) {
        long t0 = System.currentTimeMillis();
        try {
            o.wait(5_000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        if (System.currentTimeMillis() - t0 > 4_000) {
            throw new RuntimeException("Timeout!");
        }
    }
}