Repository: reactor/reactor-core
Issue: EmitterProcessor queue not cleared on immediate cancellation

`EmitterProcessor#remove` has support to discard the buffer once the last subscriber is gone. However, it returns early if the subscribers array is already `EMPTY`. For cases where the subscription is immediately cancelled (e.g. `.take(0)`) this may leak the buffered elements. 

Here is a repro case:
```
@Test
void shouldClearBufferOnCancellation() {
  BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();

  Many<Integer> sink = Sinks.many().unicast().onBackpressureBuffer(queue);

  sink.tryEmitNext(1);
  sink.tryEmitNext(2);
  sink.tryEmitNext(3);

  assertThat(queue).hasSize(3);

  sink.asFlux().take(0).blockLast();

  assertThat(queue).isEmpty();
}
```

As a workaround `.takeWhile(x -> false)` does make the test pass for example.

Changed files:
reactor-core/src/main/java/reactor/core/publisher/Sinks.java
reactor-core/src/test/java/reactor/test/MemoryUtils.java
reactor-core/src/test/java/reactor/core/publisher/SinkManyEmitterProcessorTest.java
reactor-core/src/main/java/reactor/core/publisher/UnicastProcessor.java
reactor-core/src/main/java/reactor/core/publisher/EmitterProcessor.java
reactor-core/src/main/java/reactor/core/publisher/SinksSpecs.java
reactor-core/src/test/java/reactor/core/publisher/OnDiscardShouldNotLeakTest.java

diff --git a/reactor-core/src/main/java/reactor/core/publisher/EmitterProcessor.java b/reactor-core/src/main/java/reactor/core/publisher/EmitterProcessor.java
index cedbf7d68..47ffd686e 100644
--- a/reactor-core/src/main/java/reactor/core/publisher/EmitterProcessor.java
+++ b/reactor-core/src/main/java/reactor/core/publisher/EmitterProcessor.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,6 +21,7 @@ import java.util.Queue;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import java.util.function.Consumer;
 import java.util.stream.Stream;
 
 import org.reactivestreams.Publisher;
@@ -33,6 +34,8 @@ import reactor.core.Exceptions;
 import reactor.core.Fuseable;
 import reactor.core.Scannable;
 import reactor.core.publisher.Sinks.EmitResult;
+import reactor.util.Logger;
+import reactor.util.Loggers;
 import reactor.util.annotation.Nullable;
 import reactor.util.concurrent.Queues;
 import reactor.util.context.Context;
@@ -70,6 +73,8 @@ import static reactor.core.publisher.FluxPublish.PublishSubscriber.TERMINATED;
 public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements InternalManySink<T>,
 	Sinks.ManyWithUpstream<T> {
 
+	static final Logger log = Loggers.getLogger(EmitterProcessor.class);
+
 	@SuppressWarnings("rawtypes")
 	static final FluxPublish.PubSubInner[] EMPTY = new FluxPublish.PublishInner[0];
 
@@ -133,13 +138,32 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 	 */
 	@Deprecated
 	public static <E> EmitterProcessor<E> create(int bufferSize, boolean autoCancel) {
-		return new EmitterProcessor<>(autoCancel, bufferSize);
+		return new EmitterProcessor<>(autoCancel, bufferSize, null);
+	}
+
+	/**
+	 * Create a new {@link EmitterProcessor} using the provided backlog size and auto-cancellation.
+	 *
+	 * @param <E> Type of processed signals
+	 * @param bufferSize the internal buffer size to hold signals
+	 * @param autoCancel automatically cancel
+	 *
+	 * @return a fresh processor
+	 * @deprecated use {@link Sinks.MulticastSpec#onBackpressureBuffer(int, boolean) Sinks.many().multicast().onBackpressureBuffer(bufferSize, autoCancel)}
+	 * (or the unsafe variant if you're sure about external synchronization). To be removed in 3.5.
+	 */
+	@Deprecated
+	public static <E> EmitterProcessor<E> create(int bufferSize, boolean autoCancel, Consumer<? super E> onDiscardHook) {
+		return new EmitterProcessor<>(autoCancel, bufferSize, onDiscardHook);
 	}
 
 	final int prefetch;
 
 	final boolean autoCancel;
 
+	@Nullable
+	final Consumer<? super T> onDiscard;
+
 	volatile Subscription s;
 	@SuppressWarnings("rawtypes")
 	static final AtomicReferenceFieldUpdater<EmitterProcessor, Subscription> S =
@@ -182,12 +206,13 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 					Throwable.class,
 					"error");
 
-	EmitterProcessor(boolean autoCancel, int prefetch) {
+	EmitterProcessor(boolean autoCancel, int prefetch, @Nullable Consumer<? super T> onDiscard) {
 		if (prefetch < 1) {
 			throw new IllegalArgumentException("bufferSize must be strictly positive, " + "was: " + prefetch);
 		}
 		this.autoCancel = autoCancel;
 		this.prefetch = prefetch;
+		this.onDiscard = onDiscard;
 		//doesn't use INIT/CANCELLED distinction, contrary to FluxPublish)
 		//see remove()
 		SUBSCRIBERS.lazySet(this, EMPTY);
@@ -200,7 +225,12 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 
 	@Override
 	public Context currentContext() {
-		return Operators.multiSubscribersContext(subscribers);
+		if (onDiscard != null) {
+			return Operators.enableOnDiscard(Operators.multiSubscribersContext(subscribers), onDiscard);
+		}
+		else {
+			return Operators.multiSubscribersContext(subscribers);
+		}
 	}
 
 
@@ -213,9 +243,18 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 			done = true;
 			CancellationException detachException = new CancellationException("the ManyWithUpstream sink had a Subscription to an upstream which has been manually cancelled");
 			if (ERROR.compareAndSet(EmitterProcessor.this, null, detachException)) {
+				if (WIP.getAndIncrement(this) != 0) {
+					return true;
+				}
 				Queue<T> q = queue;
 				if (q != null) {
-					q.clear();
+					Consumer<? super T> hook = this.onDiscard;
+					if (hook != null) {
+						discardQueue(q, hook);
+					}
+					else {
+						q.clear();
+					}
 				}
 				for (FluxPublish.PubSubInner<T> inner : terminate()) {
 					inner.actual.onError(detachException);
@@ -250,7 +289,7 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 			if (inner.isCancelled()) {
 				remove(inner);
 			}
-			drain();
+			drain(null);
 		}
 		else {
 			Throwable e = error;
@@ -275,7 +314,7 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 			return EmitResult.FAIL_TERMINATED;
 		}
 		done = true;
-		drain();
+		drain(null);
 		return EmitResult.OK;
 	}
 
@@ -292,7 +331,7 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 		}
 		if (Exceptions.addThrowable(ERROR, this, t)) {
 			done = true;
-			drain();
+			drain(null);
 			return EmitResult.OK;
 		}
 		else {
@@ -303,7 +342,7 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 	@Override
 	public void onNext(T t) {
 		if (sourceMode == Fuseable.ASYNC) {
-			drain();
+			drain(t);
 			return;
 		}
 		emitNext(t, Sinks.EmitFailureHandler.FAIL_FAST);
@@ -340,7 +379,8 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 		if (!q.offer(t)) {
 			return subscribers == EMPTY ? EmitResult.FAIL_ZERO_SUBSCRIBER : EmitResult.FAIL_OVERFLOW;
 		}
-		drain();
+
+		drain(t);
 		return EmitResult.OK;
 	}
 
@@ -387,7 +427,7 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 				if (m == Fuseable.SYNC) {
 					sourceMode = m;
 					queue = f;
-					drain();
+					drain(null);
 					return;
 				}
 				else if (m == Fuseable.ASYNC) {
@@ -443,8 +483,23 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 		return super.scanUnsafe(key);
 	}
 
-	final void drain() {
+	final void drain(@Nullable T dataSignalOfferedBeforeDrain) {
 		if (WIP.getAndIncrement(this) != 0) {
+			Consumer<? super T> discardHook = this.onDiscard;
+			if (dataSignalOfferedBeforeDrain != null) {
+				if (discardHook != null && isCancelled()) {
+					try {
+						discardHook.accept(dataSignalOfferedBeforeDrain);
+					}
+					catch (Throwable t) {
+						log.warn("Error in discard hook", t);
+					}
+				}
+				else if (done) {
+					Operators.onNextDropped(dataSignalOfferedBeforeDrain,
+							currentContext());
+				}
+			}
 			return;
 		}
 
@@ -458,11 +513,12 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 
 			boolean empty = q == null || q.isEmpty();
 
-			if (checkTerminated(d, empty)) {
+			if (checkTerminated(d, empty, null)) {
 				return;
 			}
 
 			FluxPublish.PubSubInner<T>[] a = subscribers;
+			Consumer<? super T> onDiscardHook = onDiscard;
 
 			if (a != EMPTY && !empty) {
 				long maxRequested = Long.MAX_VALUE;
@@ -492,9 +548,13 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 						d = true;
 						v = null;
 					}
-					if (checkTerminated(d, v == null)) {
+					empty = v == null;
+					if (checkTerminated(d, empty, v)) {
 						return;
 					}
+					if (!empty && onDiscardHook != null) {
+						discard(v, onDiscardHook);
+					}
 					if (sourceMode != Fuseable.SYNC) {
 						s.request(1);
 					}
@@ -519,7 +579,7 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 
 					empty = v == null;
 
-					if (checkTerminated(d, empty)) {
+					if (checkTerminated(d, empty, v)) {
 						return;
 					}
 
@@ -528,7 +588,7 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 						if (sourceMode == Fuseable.SYNC) {
 							//the q is empty
 							done = true;
-							checkTerminated(true, true);
+							checkTerminated(true, true, null);
 						}
 						break;
 					}
@@ -555,7 +615,7 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 			}
 			else if ( sourceMode == Fuseable.SYNC ) {
 				done = true;
-				if (checkTerminated(true, empty)) { //empty can be true if no subscriber
+				if (checkTerminated(true, empty, null)) { //empty can be true if no subscriber
 					break;
 				}
 			}
@@ -572,13 +632,23 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 		return SUBSCRIBERS.getAndSet(this, TERMINATED);
 	}
 
-	boolean checkTerminated(boolean d, boolean empty) {
+	boolean checkTerminated(boolean d, boolean empty, @Nullable T value) {
 		if (s == Operators.cancelledSubscription()) {
 			if (autoCancel) {
 				terminate();
+
 				Queue<T> q = queue;
 				if (q != null) {
-					q.clear();
+					Consumer<? super T> hook = this.onDiscard;
+					if (hook != null) {
+						if (value != null) {
+							discard(value, hook);
+						}
+						discardQueue(q, hook);
+					}
+					else {
+						q.clear();
+					}
 				}
 			}
 			return true;
@@ -588,7 +658,16 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 			if (e != null && e != Exceptions.TERMINATED) {
 				Queue<T> q = queue;
 				if (q != null) {
-					q.clear();
+					Consumer<? super T> hook = this.onDiscard;
+					if (hook != null) {
+						if (value != null) {
+							discard(value, hook);
+						}
+						discardQueue(q, hook);
+					}
+					else {
+						q.clear();
+					}
 				}
 				for (FluxPublish.PubSubInner<T> inner : terminate()) {
 					inner.actual.onError(e);
@@ -605,6 +684,31 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 		return false;
 	}
 
+	static <T> void discardQueue(Queue<T> q, Consumer<? super T> hook) {
+		for (; ; ) {
+			T toDiscard = q.poll();
+			if (toDiscard == null) {
+				break;
+			}
+
+			try {
+				hook.accept(toDiscard);
+			}
+			catch (Throwable t) {
+				log.warn("Error while discarding a queue element, continuing with next queue element", t);
+			}
+		}
+	}
+
+	static <T> void discard(T value, Consumer<? super T> hook) {
+		try {
+			hook.accept(value);
+		}
+		catch (Throwable t) {
+			log.warn("Error while discarding a queue element, continuing with next queue element", t);
+		}
+	}
+
 	final boolean add(EmitterInner<T> inner) {
 		for (; ; ) {
 			FluxPublish.PubSubInner<T>[] a = subscribers;
@@ -624,7 +728,27 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 	final void remove(FluxPublish.PubSubInner<T> inner) {
 		for (; ; ) {
 			FluxPublish.PubSubInner<T>[] a = subscribers;
-			if (a == TERMINATED || a == EMPTY) {
+			if (a == EMPTY) {
+				// means cancelled without adding
+				if (autoCancel && Operators.terminate(S, this)) {
+					if (WIP.getAndIncrement(this) != 0) {
+						return;
+					}
+					terminate();
+					Queue<T> q = queue;
+					if (q != null) {
+						Consumer<? super T> hook = this.onDiscard;
+						if (hook != null) {
+							discardQueue(q, hook);
+						}
+						else {
+							q.clear();
+						}
+					}
+				}
+				return;
+			}
+			else if (a == TERMINATED) {
 				return;
 			}
 			int n = a.length;
@@ -659,7 +783,13 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 					terminate();
 					Queue<T> q = queue;
 					if (q != null) {
-						q.clear();
+						Consumer<? super T> hook = this.onDiscard;
+						if (hook != null) {
+							discardQueue(q, hook);
+						}
+						else {
+							q.clear();
+						}
 					}
 				}
 				return;
@@ -683,13 +813,13 @@ public final class EmitterProcessor<T> extends FluxProcessor<T, T> implements In
 
 		@Override
 		void drainParent() {
-			parent.drain();
+			parent.drain(null);
 		}
 
 		@Override
 		void removeAndDrainParent() {
 			parent.remove(this);
-			parent.drain();
+			parent.drain(null);
 		}
 	}
 
diff --git a/reactor-core/src/main/java/reactor/core/publisher/Sinks.java b/reactor-core/src/main/java/reactor/core/publisher/Sinks.java
index e9183e4ba..16ceea978 100644
--- a/reactor-core/src/main/java/reactor/core/publisher/Sinks.java
+++ b/reactor-core/src/main/java/reactor/core/publisher/Sinks.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2020-2023 VMware Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@ package reactor.core.publisher;
 
 import java.time.Duration;
 import java.util.Queue;
+import java.util.function.Consumer;
 
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
@@ -420,6 +421,31 @@ public final class Sinks {
 		 * @param autoCancel should the sink fully shutdowns (not publishing anymore) when the last subscriber cancels
 		 */
 		<T> ManyWithUpstream<T> multicastOnBackpressureBuffer(int bufferSize, boolean autoCancel);
+
+		/**
+		 * A {@link Sinks.ManyWithUpstream} with the following characteristics:
+		 * <ul>
+		 *     <li>Multicast</li>
+		 *     <li>Without {@link Subscriber}: warm up. Remembers up to {@code bufferSize}
+		 *     elements pushed via {@link Many#tryEmitNext(Object)} before the first {@link Subscriber} is registered.</li>
+		 *     <li>Backpressure : this sink honors downstream demand by conforming to the lowest demand in case
+		 *     of multiple subscribers.<br>If the difference between multiple subscribers is too high compared to {@code bufferSize}:
+		 *          <ul><li>{@link Many#tryEmitNext(Object) tryEmitNext} will return {@link EmitResult#FAIL_OVERFLOW}</li>
+		 *          <li>{@link Many#emitNext(Object, Sinks.EmitFailureHandler) emitNext} will terminate the sink by {@link Many#emitError(Throwable, Sinks.EmitFailureHandler) emitting}
+		 *          an {@link Exceptions#failWithOverflow() overflow error}.</li></ul>
+		 *     </li>
+		 *     <li>Replaying: No replay of values seen by earlier subscribers. Only forwards to a {@link Subscriber}
+		 *     the elements that have been pushed to the sink AFTER this subscriber was subscribed, or elements
+		 *     that have been buffered due to backpressure/warm up.</li>
+		 * </ul>
+		 * <p>
+		 * <img class="marble" src="doc-files/marbles/sinkWarmup.svg" alt="">
+		 *
+		 * @param bufferSize the maximum queue size
+		 * @param autoCancel should the sink fully shutdowns (not publishing anymore) when the last subscriber cancels
+		 * @param onDiscardHook should be called when values from queue are cleared
+		 */
+		<T> ManyWithUpstream<T> multicastOnBackpressureBuffer(int bufferSize, boolean autoCancel, Consumer<? super T> onDiscardHook);
 	}
 
 	/**
@@ -557,6 +583,31 @@ public final class Sinks {
 		 */
 		<T> Sinks.Many<T> onBackpressureBuffer(int bufferSize, boolean autoCancel);
 
+		/**
+		 * A {@link Sinks.Many} with the following characteristics:
+		 * <ul>
+		 *     <li>Multicast</li>
+		 *     <li>Without {@link Subscriber}: warm up. Remembers up to {@code bufferSize}
+		 *     elements pushed via {@link Many#tryEmitNext(Object)} before the first {@link Subscriber} is registered.</li>
+		 *     <li>Backpressure : this sink honors downstream demand by conforming to the lowest demand in case
+		 *     of multiple subscribers.<br>If the difference between multiple subscribers is too high compared to {@code bufferSize}:
+		 *          <ul><li>{@link Many#tryEmitNext(Object) tryEmitNext} will return {@link EmitResult#FAIL_OVERFLOW}</li>
+		 *          <li>{@link Many#emitNext(Object, Sinks.EmitFailureHandler) emitNext} will terminate the sink by {@link Many#emitError(Throwable, Sinks.EmitFailureHandler) emitting}
+		 *          an {@link Exceptions#failWithOverflow() overflow error}.</li></ul>
+		 *     </li>
+		 *     <li>Replaying: No replay of values seen by earlier subscribers. Only forwards to a {@link Subscriber}
+		 *     the elements that have been pushed to the sink AFTER this subscriber was subscribed, or elements
+		 *     that have been buffered due to backpressure/warm up.</li>
+		 * </ul>
+		 * <p>
+		 * <img class="marble" src="doc-files/marbles/sinkWarmup.svg" alt="">
+		 *
+		 * @param bufferSize the maximum queue size
+		 * @param autoCancel should the sink fully shutdowns (not publishing anymore) when the last subscriber cancels
+		 * @param onDiscardHook should be called when values from queue are cleared
+		 */
+		<T> Sinks.Many<T> onBackpressureBuffer(int bufferSize, boolean autoCancel, Consumer<? super T> onDiscardHook);
+
 		/**
 		 A {@link Sinks.Many} with the following characteristics:
 		 * <ul>
diff --git a/reactor-core/src/main/java/reactor/core/publisher/SinksSpecs.java b/reactor-core/src/main/java/reactor/core/publisher/SinksSpecs.java
index 33d6f45ce..cb4d48f70 100644
--- a/reactor-core/src/main/java/reactor/core/publisher/SinksSpecs.java
+++ b/reactor-core/src/main/java/reactor/core/publisher/SinksSpecs.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2020-2023 VMware Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,6 +20,7 @@ import java.time.Duration;
 import java.util.Queue;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import java.util.function.Consumer;
 
 import reactor.core.Disposable;
 import reactor.core.publisher.Sinks.Empty;
@@ -109,17 +110,22 @@ final class SinksSpecs {
 
 		@Override
 		public <T> Sinks.Many<T> onBackpressureBuffer() {
-			return new SinkManyEmitterProcessor<>(true, Queues.SMALL_BUFFER_SIZE);
+			return new SinkManyEmitterProcessor<>(true, Queues.SMALL_BUFFER_SIZE, null);
 		}
 
 		@Override
 		public <T> Sinks.Many<T> onBackpressureBuffer(int bufferSize) {
-			return new SinkManyEmitterProcessor<>(true, bufferSize);
+			return new SinkManyEmitterProcessor<>(true, bufferSize, null);
 		}
 
 		@Override
 		public <T> Sinks.Many<T> onBackpressureBuffer(int bufferSize, boolean autoCancel) {
-			return new SinkManyEmitterProcessor<>(autoCancel, bufferSize);
+			return new SinkManyEmitterProcessor<>(autoCancel, bufferSize, null);
+		}
+
+		@Override
+		public <T> Sinks.Many<T> onBackpressureBuffer(int bufferSize, boolean autoCancel, Consumer<? super T> hook) {
+			return new EmitterProcessor<>(autoCancel, bufferSize, hook);
 		}
 
 		@Override
@@ -179,12 +185,17 @@ final class SinksSpecs {
 
 		@Override
 		public <T> Sinks.ManyWithUpstream<T> multicastOnBackpressureBuffer() {
-			return new SinkManyEmitterProcessor<>(true, Queues.SMALL_BUFFER_SIZE);
+			return new SinkManyEmitterProcessor<>(true, Queues.SMALL_BUFFER_SIZE, null);
 		}
 
 		@Override
 		public <T> Sinks.ManyWithUpstream<T> multicastOnBackpressureBuffer(int bufferSize, boolean autoCancel) {
-			return new SinkManyEmitterProcessor<>(autoCancel, bufferSize);
+			return new SinkManyEmitterProcessor<>(autoCancel, bufferSize, null);
+		}
+
+		@Override
+		public <T> Sinks.ManyWithUpstream<T> multicastOnBackpressureBuffer(int bufferSize, boolean autoCancel, Consumer<? super T> hook) {
+			return new EmitterProcessor<>(autoCancel, bufferSize, hook);
 		}
 	}
 
@@ -252,6 +263,13 @@ final class SinksSpecs {
 			return wrapMany(new SinkManyEmitterProcessor<>(autoCancel, bufferSize));
 		}
 
+		@Override
+		public <T> Many<T> onBackpressureBuffer(int bufferSize, boolean autoCancel, Consumer<? super T> onDiscardHook) {
+			@SuppressWarnings("deprecation") // EmitterProcessor will be removed in 3.5.
+			final EmitterProcessor<T> original = EmitterProcessor.create(bufferSize, autoCancel, onDiscardHook);
+			return wrapMany(original);
+		}
+
 		@Override
 		public <T> Many<T> directAllOrNothing() {
 			final SinkManyBestEffort<T> original = SinkManyBestEffort.createAllOrNothing();
@@ -351,6 +369,7 @@ final class SinksSpecs {
 		}
 
 		@Override
+
 		public <T> Many<T> onBackpressureBuffer(Queue<T> queue) {
 			final SinkManyUnicast<T> original = SinkManyUnicast.create(queue);
 			return wrapMany(original);
diff --git a/reactor-core/src/main/java/reactor/core/publisher/UnicastProcessor.java b/reactor-core/src/main/java/reactor/core/publisher/UnicastProcessor.java
index d32f9e7f1..fc8556136 100644
--- a/reactor-core/src/main/java/reactor/core/publisher/UnicastProcessor.java
+++ b/reactor-core/src/main/java/reactor/core/publisher/UnicastProcessor.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -530,9 +530,9 @@ public final class UnicastProcessor<T> extends FluxProcessor<T, T>
 			this.actual = actual;
 			if (cancelled) {
 				this.hasDownstream = false;
-			} else {
-				drain(null);
 			}
+
+			drain(null);
 		} else {
 			Operators.error(actual, new IllegalStateException("UnicastProcessor " +
 					"allows only a single Subscriber"));
@@ -556,7 +556,7 @@ public final class UnicastProcessor<T> extends FluxProcessor<T, T>
 
 		doTerminate();
 
-		if (WIP.getAndIncrement(this) == 0) {
+		if (actual != null && WIP.getAndIncrement(this) == 0) {
 			if (!outputFused) {
 				// discard MUST be happening only and only if there is no racing on elements consumption
 				// which is guaranteed by the WIP guard here in case non-fused output
diff --git a/reactor-core/src/test/java/reactor/core/publisher/OnDiscardShouldNotLeakTest.java b/reactor-core/src/test/java/reactor/core/publisher/OnDiscardShouldNotLeakTest.java
index f1cc13d99..acffdc76b 100644
--- a/reactor-core/src/test/java/reactor/core/publisher/OnDiscardShouldNotLeakTest.java
+++ b/reactor-core/src/test/java/reactor/core/publisher/OnDiscardShouldNotLeakTest.java
@@ -172,6 +172,12 @@ public class OnDiscardShouldNotLeakTest {
 			DiscardScenario.sinkSource("unicastSink",  Sinks.unsafe().many().unicast()::onBackpressureBuffer, null),
 			DiscardScenario.sinkSource("unicastSinkAndPublishOn",  Sinks.unsafe().many().unicast()::onBackpressureBuffer,
 					f -> f.publishOn(Schedulers.immediate())),
+			DiscardScenario.sinkSource("multicastBufferSink",
+					() -> Sinks.unsafe().many().multicast().onBackpressureBuffer(32, true, Tracked::release),
+					null),
+			DiscardScenario.sinkSource("multicastBufferSinkAndPublishOn",
+					() -> Sinks.unsafe().many().multicast().onBackpressureBuffer(32, true, Tracked::release),
+					f -> f.publishOn(Schedulers.immediate())),
 			DiscardScenario.fluxSource("singleOrEmpty", f -> f.singleOrEmpty().onErrorReturn(Tracked.RELEASED)),
 			DiscardScenario.fluxSource("collect", f -> f.collect(ArrayList::new, ArrayList::add)
 			                                               .doOnSuccess(l -> l.forEach(Tracked::safeRelease))
@@ -656,7 +662,11 @@ public class OnDiscardShouldNotLeakTest {
 				Sinks.Many<Tracked> sink = sinksManySupplier.get();
 				//noinspection CallingSubscribeInNonBlockingScope
 				main.flux().subscribe(
-						v -> sink.emitNext(v, FAIL_FAST),
+						v -> {
+							if (sink.tryEmitNext(v) != Sinks.EmitResult.OK) {
+								v.release();
+							}
+						},
 						e -> sink.emitError(e, FAIL_FAST),
 						() -> sink.emitComplete(FAIL_FAST));
 				Flux<Tracked> finalFlux = sink.asFlux();
diff --git a/reactor-core/src/test/java/reactor/core/publisher/SinkManyEmitterProcessorTest.java b/reactor-core/src/test/java/reactor/core/publisher/SinkManyEmitterProcessorTest.java
index 449d47892..9b72087e5 100644
--- a/reactor-core/src/test/java/reactor/core/publisher/SinkManyEmitterProcessorTest.java
+++ b/reactor-core/src/test/java/reactor/core/publisher/SinkManyEmitterProcessorTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -26,6 +26,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Stream;
 
+import org.assertj.core.api.Assertions;
 import org.awaitility.Awaitility;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
@@ -40,6 +41,7 @@ import reactor.core.Scannable;
 import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 import reactor.test.AutoDisposingExtension;
+import reactor.test.MemoryUtils;
 import reactor.test.StepVerifier;
 import reactor.test.publisher.TestPublisher;
 import reactor.test.subscriber.AssertSubscriber;
@@ -62,6 +64,21 @@ class SinkManyEmitterProcessorTest {
 
 	@RegisterExtension
 	AutoDisposingExtension afterTest = new AutoDisposingExtension();
+	//https://github.com/reactor/reactor-core/issues/3359
+	@Test
+	void shouldClearBufferOnCancellation() {
+		MemoryUtils.OffHeapDetector tracker = new MemoryUtils.OffHeapDetector();
+		EmitterProcessor<MemoryUtils.Tracked> source = EmitterProcessor.create(32, true, MemoryUtils.Tracked::release);
+
+		Assertions.assertThat(source.tryEmitNext(tracker.track(1))).isEqualTo(Sinks.EmitResult.OK);
+		Assertions.assertThat(source.tryEmitNext(tracker.track(2))).isEqualTo(Sinks.EmitResult.OK);
+		Assertions.assertThat(source.tryEmitNext(tracker.track(3))).isEqualTo(Sinks.EmitResult.OK);
+
+		source.take(0).blockLast();
+
+		tracker.assertNoLeaks();
+	}
+
 
 	@Test
 	void smokeTestManyWithUpstream() {
@@ -862,7 +879,7 @@ class SinkManyEmitterProcessorTest {
 		AssertSubscriber<Object> testSubscriber1 = new AssertSubscriber<>(Context.of("key", "value1"));
 		AssertSubscriber<Object> testSubscriber2 = new AssertSubscriber<>(Context.of("key", "value2"));
 
-		SinkManyEmitterProcessor<Object> sinkManyEmitterProcessor = new SinkManyEmitterProcessor<>(false, 1);
+		SinkManyEmitterProcessor<Object> sinkManyEmitterProcessor = new SinkManyEmitterProcessor<>(false, 1, null);
 		sinkManyEmitterProcessor.subscribe(testSubscriber1);
 		sinkManyEmitterProcessor.subscribe(testSubscriber2);
 
diff --git a/reactor-core/src/test/java/reactor/test/MemoryUtils.java b/reactor-core/src/test/java/reactor/test/MemoryUtils.java
index 55223eeb3..e32ba8420 100644
--- a/reactor-core/src/test/java/reactor/test/MemoryUtils.java
+++ b/reactor-core/src/test/java/reactor/test/MemoryUtils.java
@@ -18,6 +18,7 @@ package reactor.test;
 
 import java.lang.ref.PhantomReference;
 import java.lang.ref.ReferenceQueue;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.LinkedList;
 import java.util.List;
@@ -170,6 +171,8 @@ public class MemoryUtils {
 		 */
 		public static final Tracked RELEASED = new Tracked("RELEASED", true);
 
+		final List<String> touches = new ArrayList<>();
+
 		/**
 		 * Check if an arbitrary object is a {@link Tracked}, and if so release it.
 		 *
@@ -210,6 +213,10 @@ public class MemoryUtils {
 	        set(true);
 	    }
 
+		public void touch(String msg) {
+			touches.add(msg);
+		}
+
 		/**
 		 * Check if this {@link Tracked} object has been released.
 		 *
@@ -221,6 +228,10 @@ public class MemoryUtils {
 
 	    @Override
 	    public boolean equals(Object o) {
+		    if (o instanceof String) {
+			    touch(o.toString());
+				return false;
+		    }
 	        if (this == o) return true;
 	        if (o == null || getClass() != o.getClass()) return false;
 
@@ -237,10 +248,16 @@ public class MemoryUtils {
 	    //NOTE: AssertJ has a special representation of AtomicBooleans, so we override it in AssertionsUtils
 	    @Override
 	    public String toString() {
-	        return "Tracked{" +
-	                " id=" + identifier +
-	                " released=" + get() +
-	                " }";
+	        return get()
+			        ? "Tracked{" +
+		                " id=" + identifier +
+		                " released=" + get() +
+		                " }"
+			        : "Tracked{" +
+				        " id=" + identifier +
+				        " released=" + get() +
+				        " touches=" + touches +
+				        " }";
 	    }
 	}
 }