diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 05282e7..2fd4c57 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -52,6 +52,7 @@
 mockito-core = { module = "org.mockito:mockito-core", version.ref = "mockito" }
 pulsar-client-api = { module = "org.apache.pulsar:pulsar-client-api", version.ref = "pulsar" }
 pulsar-client-shaded = { module = "org.apache.pulsar:pulsar-client", version.ref = "pulsar" }
+pulsar-client-all = { module = "org.apache.pulsar:pulsar-client-all", version.ref = "pulsar" }
 rat-gradle = { module = "org.nosphere.apache:creadur-rat-gradle", version.ref = "rat-gradle" }
 reactor-core = { module = "io.projectreactor:reactor-core", version.ref = "reactor" }
 reactor-test = { module = "io.projectreactor:reactor-test", version.ref = "reactor" }
diff --git a/pulsar-client-reactive-adapter/build.gradle b/pulsar-client-reactive-adapter/build.gradle
index 984775a..37e6774 100644
--- a/pulsar-client-reactive-adapter/build.gradle
+++ b/pulsar-client-reactive-adapter/build.gradle
@@ -35,6 +35,7 @@
 	testImplementation libs.bundles.log4j
 	testImplementation libs.mockito.core
 
+	intTestImplementation libs.pulsar.client.all
 	intTestImplementation project(':pulsar-client-reactive-producer-cache-caffeine')
 	intTestImplementation project(path: ':pulsar-client-reactive-producer-cache-caffeine-shaded', configuration: 'shadow')
 	intTestImplementation libs.junit.jupiter
diff --git a/pulsar-client-reactive-adapter/src/intTest/java/org/apache/pulsar/reactive/client/adapter/ReactiveMessagePipelineE2ETests.java b/pulsar-client-reactive-adapter/src/intTest/java/org/apache/pulsar/reactive/client/adapter/ReactiveMessagePipelineE2ETests.java
index 6c8dada..a93988d 100644
--- a/pulsar-client-reactive-adapter/src/intTest/java/org/apache/pulsar/reactive/client/adapter/ReactiveMessagePipelineE2ETests.java
+++ b/pulsar-client-reactive-adapter/src/intTest/java/org/apache/pulsar/reactive/client/adapter/ReactiveMessagePipelineE2ETests.java
@@ -36,8 +36,11 @@
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
+import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.policies.data.SubscriptionStats;
+import org.apache.pulsar.common.policies.data.TopicStats;
 import org.apache.pulsar.reactive.client.api.MessageSpec;
 import org.apache.pulsar.reactive.client.api.MessageSpecBuilder;
 import org.apache.pulsar.reactive.client.api.ReactiveMessagePipeline;
@@ -94,6 +97,38 @@
 		}
 	}
 
+	@Test
+	void shouldSupportWaitingForConsumingToStartAndStop() throws Exception {
+		try (PulsarClient pulsarClient = SingletonPulsarContainer.createPulsarClient();
+				PulsarAdmin pulsarAdmin = SingletonPulsarContainer.createPulsarAdmin()) {
+			String topicName = "test" + UUID.randomUUID();
+			ReactivePulsarClient reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient);
+			ReactiveMessagePipeline pipeline = reactivePulsarClient.messageConsumer(Schema.STRING)
+				.subscriptionName("sub")
+				.topic(topicName)
+				.build()
+				.messagePipeline()
+				.messageHandler((message) -> Mono.empty())
+				.build()
+				.start();
+
+			// wait for consuming to start
+			pipeline.untilStarted().block(Duration.ofSeconds(5));
+			// there should be an existing subscription
+			List<String> subscriptions = pulsarAdmin.topics().getSubscriptions(topicName);
+			assertThat(subscriptions).as("subscription should be created").contains("sub");
+
+			// stop the pipeline
+			pipeline.stop();
+			// and wait for it to stop
+			pipeline.untilStopped().block(Duration.ofSeconds(5));
+			// there should be no consumers
+			TopicStats topicStats = pulsarAdmin.topics().getStats(topicName);
+			SubscriptionStats subStats = topicStats.getSubscriptions().get("sub");
+			assertThat(subStats.getConsumers()).isEmpty();
+		}
+	}
+
 	@ParameterizedTest
 	@EnumSource(MessageOrderScenario.class)
 	void shouldRetainMessageOrder(MessageOrderScenario messageOrderScenario) throws Exception {
diff --git a/pulsar-client-reactive-adapter/src/intTest/java/org/apache/pulsar/reactive/client/adapter/SingletonPulsarContainer.java b/pulsar-client-reactive-adapter/src/intTest/java/org/apache/pulsar/reactive/client/adapter/SingletonPulsarContainer.java
index 431e448..d53e185 100644
--- a/pulsar-client-reactive-adapter/src/intTest/java/org/apache/pulsar/reactive/client/adapter/SingletonPulsarContainer.java
+++ b/pulsar-client-reactive-adapter/src/intTest/java/org/apache/pulsar/reactive/client/adapter/SingletonPulsarContainer.java
@@ -19,6 +19,7 @@
 
 package org.apache.pulsar.reactive.client.adapter;
 
+import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.testcontainers.containers.PulsarContainer;
@@ -44,6 +45,12 @@
 			.build();
 	}
 
+	static PulsarAdmin createPulsarAdmin() throws PulsarClientException {
+		return PulsarAdmin.builder()
+			.serviceHttpUrl(SingletonPulsarContainer.PULSAR_CONTAINER.getHttpServiceUrl())
+			.build();
+	}
+
 	static DockerImageName getPulsarImage() {
 		return DockerImageName.parse("apachepulsar/pulsar:4.0.4");
 	}
diff --git a/pulsar-client-reactive-adapter/src/main/java/org/apache/pulsar/reactive/client/internal/adapter/ReactiveConsumerAdapter.java b/pulsar-client-reactive-adapter/src/main/java/org/apache/pulsar/reactive/client/internal/adapter/ReactiveConsumerAdapter.java
index 8e89a09..e66ba28 100644
--- a/pulsar-client-reactive-adapter/src/main/java/org/apache/pulsar/reactive/client/internal/adapter/ReactiveConsumerAdapter.java
+++ b/pulsar-client-reactive-adapter/src/main/java/org/apache/pulsar/reactive/client/internal/adapter/ReactiveConsumerAdapter.java
@@ -25,6 +25,7 @@
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.ConsumerBuilder;
 import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.reactive.client.internal.api.InternalConsumerListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
@@ -45,12 +46,20 @@
 	}
 
 	private Mono<Consumer<T>> createConsumerMono() {
-		return AdapterImplementationFactory.adaptPulsarFuture(
-				() -> this.consumerBuilderFactory.apply(this.pulsarClientSupplier.get()).subscribeAsync());
+		return Mono.deferContextual((contextView) -> AdapterImplementationFactory
+			.adaptPulsarFuture(
+					() -> this.consumerBuilderFactory.apply(this.pulsarClientSupplier.get()).subscribeAsync())
+			.doOnSuccess((consumer) -> contextView.<InternalConsumerListener>getOrEmpty(InternalConsumerListener.class)
+				.ifPresent((listener) -> listener.onConsumerCreated(consumer))));
 	}
 
 	private Mono<Void> closeConsumer(Consumer<?> consumer) {
-		return Mono.fromFuture(consumer::closeAsync).doOnSuccess((__) -> this.LOG.info("Consumer closed {}", consumer));
+		return Mono.deferContextual((contextView) -> Mono.fromFuture(consumer::closeAsync).doFinally((signalType) -> {
+			this.LOG.info("Consumer closed {}", consumer);
+			contextView.<InternalConsumerListener>getOrEmpty(InternalConsumerListener.class)
+				.ifPresent((listener) -> listener.onConsumerClosed(consumer));
+		}));
+
 	}
 
 	<R> Mono<R> usingConsumer(Function<Consumer<T>, Mono<R>> usingConsumerAction) {
diff --git a/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/api/ReactiveMessagePipeline.java b/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/api/ReactiveMessagePipeline.java
index 47804a5..9613293 100644
--- a/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/api/ReactiveMessagePipeline.java
+++ b/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/api/ReactiveMessagePipeline.java
@@ -19,20 +19,26 @@
 
 package org.apache.pulsar.reactive.client.api;
 
+import reactor.core.publisher.Mono;
+
 /**
  * Reactive message pipeline interface.
  */
 public interface ReactiveMessagePipeline extends AutoCloseable {
 
 	/**
-	 * Starts the reactive pipeline.
-	 * @return the pipeline
+	 * Starts the reactive pipeline asynchronously.
+	 * @return the pipeline instance
+	 * @see #untilStarted() For returning a reactive publisher (Mono) that completes after
+	 * the pipeline has actually started.
 	 */
 	ReactiveMessagePipeline start();
 
 	/**
-	 * Stops the reactive pipeline.
-	 * @return the reactive pipeline
+	 * Stops the reactive pipeline asynchronously.
+	 * @return the pipeline instance
+	 * @see #untilStopped() For returning a reactive publisher (Mono) that completes after
+	 * the pipeline has actually stopped.
 	 */
 	ReactiveMessagePipeline stop();
 
@@ -43,11 +49,54 @@
 	boolean isRunning();
 
 	/**
-	 * Closes the reactive pipeline.
+	 * Closes the reactive pipeline asynchronously without waiting for shutdown
+	 * completion.
 	 * @throws Exception if an error occurs
 	 */
 	default void close() throws Exception {
 		stop();
 	}
 
+	/**
+	 * <p>
+	 * Returns a reactive publisher (Mono) that completes after the pipeline has
+	 * successfully subscribed to the input topic(s) and started consuming messages for
+	 * the first time after pipeline creation. This method is not intended to be used
+	 * after a pipeline restarts following failure. Use this method to wait for consumer
+	 * and Pulsar subscription creation. This helps avoid race conditions when sending
+	 * messages immediately after the pipeline starts.
+	 * </p>
+	 * <p>
+	 * The {@link #start()} method must be called before invoking this method.
+	 * </p>
+	 * <p>
+	 * To wait for the operation to complete synchronously, it is necessary to call
+	 * {@link Mono#block()} on the returned Mono.
+	 * </p>
+	 * @return a Mono that completes after the pipeline has created its underlying Pulsar
+	 * consumer
+	 */
+	default Mono<Void> untilStarted() {
+		return Mono.empty();
+	}
+
+	/**
+	 * <p>
+	 * Returns a reactive publisher (Mono) that completes after the pipeline has closed
+	 * the underlying Pulsar consumer and stopped consuming new messages.
+	 * </p>
+	 * <p>
+	 * The {@link #stop()} method must be called before invoking this method.
+	 * </p>
+	 * <p>
+	 * To wait for the operation to complete synchronously, it is necessary to call
+	 * {@link Mono#block()} on the returned Mono.
+	 * </p>
+	 * @return a Mono that completes when the pipeline has closed the underlying Pulsar
+	 * consumer
+	 */
+	default Mono<Void> untilStopped() {
+		return Mono.empty();
+	}
+
 }
diff --git a/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/DefaultReactiveMessagePipeline.java b/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/DefaultReactiveMessagePipeline.java
index cd94f19..eefdcf2 100644
--- a/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/DefaultReactiveMessagePipeline.java
+++ b/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/DefaultReactiveMessagePipeline.java
@@ -21,6 +21,7 @@
 
 import java.time.Duration;
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiConsumer;
 import java.util.function.Function;
@@ -67,6 +68,10 @@
 
 	private final MessageGroupingFunction groupingFunction;
 
+	private final AtomicReference<InternalConsumerListenerImpl> consumerListener = new AtomicReference<>();
+
+	private final AtomicReference<CompletableFuture<Void>> pipelineStoppedFuture = new AtomicReference<>();
+
 	DefaultReactiveMessagePipeline(ReactiveMessageConsumer<T> messageConsumer,
 			Function<Message<T>, Publisher<Void>> messageHandler, BiConsumer<Message<T>, Throwable> errorLogger,
 			Retry pipelineRetrySpec, Duration handlingTimeout, Function<Mono<Void>, Publisher<Void>> transformer,
@@ -83,7 +88,14 @@
 		this.pipeline = messageConsumer.consumeMany(this::createMessageConsumer)
 			.then()
 			.transform(transformer)
-			.transform(this::decoratePipeline);
+			.transform(this::decoratePipeline)
+			.doFinally((signalType) -> {
+				CompletableFuture<Void> f = this.pipelineStoppedFuture.get();
+				if (f != null) {
+					f.complete(null);
+				}
+			})
+			.doFirst(() -> this.pipelineStoppedFuture.set(new CompletableFuture<>()));
 	}
 
 	private Mono<Void> decorateMessageHandler(Mono<Void> messageHandler) {
@@ -168,14 +180,26 @@
 		if (this.killSwitch.get() != null) {
 			throw new IllegalStateException("Message handler is already running.");
 		}
-		Disposable disposable = this.pipeline.subscribe(null, this::logError, this::logUnexpectedCompletion);
+		InternalConsumerListenerImpl consumerListener = new InternalConsumerListenerImpl();
+		Disposable disposable = this.pipeline.contextWrite(Context.of(InternalConsumerListener.class, consumerListener))
+			.subscribe(null, this::logError, this::logUnexpectedCompletion);
 		if (!this.killSwitch.compareAndSet(null, disposable)) {
 			disposable.dispose();
 			throw new IllegalStateException("Message handler was already running.");
 		}
+		this.consumerListener.set(consumerListener);
 		return this;
 	}
 
+	@Override
+	public Mono<Void> untilStarted() {
+		if (!isRunning()) {
+			throw new IllegalStateException("Pipeline isn't running. Call start first.");
+		}
+		InternalConsumerListenerImpl internalConsumerListener = this.consumerListener.get();
+		return internalConsumerListener.waitForConsumerCreated();
+	}
+
 	private void logError(Throwable throwable) {
 		LOG.error("ReactiveMessageHandler was unexpectedly terminated.", throwable);
 	}
@@ -196,8 +220,43 @@
 	}
 
 	@Override
+	public Mono<Void> untilStopped() {
+		if (isRunning()) {
+			throw new IllegalStateException("Pipeline is running. Call stop first.");
+		}
+		CompletableFuture<Void> f = this.pipelineStoppedFuture.get();
+		if (f != null) {
+			return Mono.fromFuture(f, true);
+		}
+		else {
+			return Mono.empty();
+		}
+	}
+
+	@Override
 	public boolean isRunning() {
 		return this.killSwitch.get() != null;
 	}
 
+	private static final class InternalConsumerListenerImpl implements InternalConsumerListener {
+
+		private final CompletableFuture<Void> createdFuture;
+
+		private InternalConsumerListenerImpl() {
+			this.createdFuture = new CompletableFuture<>();
+		}
+
+		@Override
+		public void onConsumerCreated(Object nativeConsumer) {
+			if (!this.createdFuture.isDone()) {
+				this.createdFuture.complete(null);
+			}
+		}
+
+		Mono<Void> waitForConsumerCreated() {
+			return Mono.fromFuture(this.createdFuture, true);
+		}
+
+	}
+
 }
diff --git a/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/InternalConsumerListener.java b/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/InternalConsumerListener.java
new file mode 100644
index 0000000..8ee8f4b
--- /dev/null
+++ b/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/InternalConsumerListener.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.pulsar.reactive.client.internal.api;
+
+/**
+ * Internal interface to signal the creation and closing of a native consumer. This is not
+ * to be intended to be used by applications.
+ */
+public interface InternalConsumerListener {
+
+	/**
+	 * Called when a new native consumer is created. This is called each time a new
+	 * consumer is created initially or as a result of a reactive pipeline retry.
+	 * @param nativeConsumer the native consumer instance
+	 */
+	default void onConsumerCreated(Object nativeConsumer) {
+		// no-op
+	}
+
+	/**
+	 * Called when a native consumer is closed. This is called each time a consumer is
+	 * closed as a result of a reactive pipeline retry or when the pipeline is closed.
+	 * @param nativeConsumer the native consumer instance
+	 */
+	default void onConsumerClosed(Object nativeConsumer) {
+		// no-op
+	}
+
+}
diff --git a/pulsar-client-reactive-api/src/test/java/org/apache/pulsar/reactive/client/api/ReactiveMessagePipelineTests.java b/pulsar-client-reactive-api/src/test/java/org/apache/pulsar/reactive/client/api/ReactiveMessagePipelineTests.java
index 3415481..f330a54 100644
--- a/pulsar-client-reactive-api/src/test/java/org/apache/pulsar/reactive/client/api/ReactiveMessagePipelineTests.java
+++ b/pulsar-client-reactive-api/src/test/java/org/apache/pulsar/reactive/client/api/ReactiveMessagePipelineTests.java
@@ -28,6 +28,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
@@ -38,6 +39,7 @@
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.internal.DefaultImplementation;
 import org.apache.pulsar.common.api.EncryptionContext;
+import org.apache.pulsar.reactive.client.internal.api.InternalConsumerListener;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
@@ -49,6 +51,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
 import static org.assertj.core.api.Assertions.assertThatNullPointerException;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 class ReactiveMessagePipelineTests {
 
@@ -165,6 +168,27 @@
 	}
 
 	@Test
+	void pipelineUntilStartedAndStopped() throws Exception {
+		int numMessages = 10;
+		Duration subscriptionDelay = Duration.ofSeconds(1);
+		TestConsumer testConsumer = new TestConsumer(numMessages, subscriptionDelay);
+		CountDownLatch latch = new CountDownLatch(numMessages);
+		Function<Message<String>, Publisher<Void>> messageHandler = (
+				message) -> Mono.empty().then().doFinally((__) -> latch.countDown());
+		ReactiveMessagePipeline pipeline = testConsumer.messagePipeline().messageHandler(messageHandler).build();
+		pipeline.start();
+		// timeout should occur since subscription delay is 1 second in TestConsumer
+		assertThatThrownBy(() -> pipeline.untilStarted().block(Duration.ofMillis(100)))
+			.isInstanceOf(IllegalStateException.class)
+			.hasCauseInstanceOf(TimeoutException.class);
+		// now wait for consuming to start
+		pipeline.untilStarted().block(Duration.ofSeconds(2));
+		assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+		// now wait for consuming to stop
+		pipeline.stop().untilStopped().block(Duration.ofSeconds(1));
+	}
+
+	@Test
 	void streamingHandler() throws Exception {
 		int numMessages = 10;
 		TestConsumer testConsumer = new TestConsumer(numMessages);
@@ -480,10 +504,17 @@
 
 		private final int numMessages;
 
+		private final Duration subscriptionDelay;
+
 		private volatile Runnable finishedCallback;
 
 		TestConsumer(int numMessages) {
+			this(numMessages, null);
+		}
+
+		TestConsumer(int numMessages, Duration subscriptionDelay) {
 			this.numMessages = numMessages;
+			this.subscriptionDelay = subscriptionDelay;
 		}
 
 		private final List<MessageId> acknowledgedMessages = new CopyOnWriteArrayList<>();
@@ -496,7 +527,10 @@
 
 		@Override
 		public <R> Flux<R> consumeMany(Function<Flux<Message<String>>, Publisher<MessageResult<R>>> messageHandler) {
-			return Flux.defer(() -> {
+			Flux<R> flux = Flux.deferContextual((contextView) -> {
+				Optional<InternalConsumerListener> internalConsumerListener = contextView
+					.getOrEmpty(InternalConsumerListener.class);
+				internalConsumerListener.ifPresent((listener) -> listener.onConsumerCreated(this));
 				Flux<Message<String>> messages = Flux.range(0, this.numMessages)
 					.map(Object::toString)
 					.map(TestMessage::new);
@@ -511,8 +545,15 @@
 					if (this.finishedCallback != null) {
 						this.finishedCallback.run();
 					}
+					internalConsumerListener.ifPresent((listener) -> listener.onConsumerClosed(this));
 				});
 			});
+			if (this.subscriptionDelay != null) {
+				return flux.delaySubscription(this.subscriptionDelay);
+			}
+			else {
+				return flux;
+			}
 		}
 
 		List<MessageId> getAcknowledgedMessages() {