diff --git a/src/DotPulsar/Abstractions/IConsumer.cs b/src/DotPulsar/Abstractions/IConsumer.cs
index ef060f4..48fe671 100644
--- a/src/DotPulsar/Abstractions/IConsumer.cs
+++ b/src/DotPulsar/Abstractions/IConsumer.cs
@@ -66,7 +66,7 @@
         bool IsFinalState(ConsumerState state);
 
         /// <summary>
-        /// Get an IAsyncEnumerable for consuming messages
+        /// Get an IAsyncEnumerable for consuming messages.
         /// </summary>
         IAsyncEnumerable<Message> Messages(CancellationToken cancellationToken = default);
 
@@ -76,6 +76,21 @@
         ValueTask Seek(MessageId messageId, CancellationToken cancellationToken = default);
 
         /// <summary>
+        /// Reset the subscription associated with this consumer to a specific message publish time using unix time in milliseconds.
+        /// </summary>
+        ValueTask Seek(ulong publishTime, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Reset the subscription associated with this consumer to a specific message publish time using an UTC DateTime.
+        /// </summary>
+        ValueTask Seek(DateTime publishTime, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Reset the subscription associated with this consumer to a specific message publish time using a DateTimeOffset.
+        /// </summary>
+        ValueTask Seek(DateTimeOffset publishTime, CancellationToken cancellationToken = default);
+
+        /// <summary>
         /// Wait for the state to change to a specific state.
         /// </summary>
         /// <returns>
diff --git a/src/DotPulsar/Abstractions/IReader.cs b/src/DotPulsar/Abstractions/IReader.cs
index a44ea9c..7b9502a 100644
--- a/src/DotPulsar/Abstractions/IReader.cs
+++ b/src/DotPulsar/Abstractions/IReader.cs
@@ -46,6 +46,26 @@
         IAsyncEnumerable<Message> Messages(CancellationToken cancellationToken = default);
 
         /// <summary>
+        /// Reset the subscription associated with this reader to a specific MessageId.
+        /// </summary>
+        ValueTask Seek(MessageId messageId, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Reset the subscription associated with this reader to a specific message publish time using unix time in milliseconds.
+        /// </summary>
+        ValueTask Seek(ulong publishTime, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Reset the subscription associated with this reader to a specific message publish time using an UTC DateTime.
+        /// </summary>
+        ValueTask Seek(DateTime publishTime, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Reset the subscription associated with this reader to a specific message publish time using a DateTimeOffset.
+        /// </summary>
+        ValueTask Seek(DateTimeOffset publishTime, CancellationToken cancellationToken = default);
+
+        /// <summary>
         /// Wait for the state to change to a specific state.
         /// </summary>
         /// <returns>
diff --git a/src/DotPulsar/Internal/Abstractions/IReaderChannel.cs b/src/DotPulsar/Internal/Abstractions/IReaderChannel.cs
index 64c2289..a63762b 100644
--- a/src/DotPulsar/Internal/Abstractions/IReaderChannel.cs
+++ b/src/DotPulsar/Internal/Abstractions/IReaderChannel.cs
@@ -14,12 +14,14 @@
 
 namespace DotPulsar.Internal.Abstractions
 {
+    using DotPulsar.Internal.PulsarApi;
     using System;
     using System.Threading;
     using System.Threading.Tasks;
 
     public interface IReaderChannel : IAsyncDisposable
     {
+        Task<CommandSuccess> Send(CommandSeek command, CancellationToken cancellationToken);
         ValueTask<Message> Receive(CancellationToken cancellationToken = default);
     }
 }
diff --git a/src/DotPulsar/Internal/Consumer.cs b/src/DotPulsar/Internal/Consumer.cs
index 33e807f..dad2416 100644
--- a/src/DotPulsar/Internal/Consumer.cs
+++ b/src/DotPulsar/Internal/Consumer.cs
@@ -124,11 +124,31 @@
         {
             ThrowIfDisposed();
 
-            var seek = new CommandSeek
-            {
-                MessageId = messageId.Data
-            };
+            var seek = new CommandSeek { MessageId = messageId.Data };
+            _ = await _executor.Execute(() => _channel.Send(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
+        }
 
+        public async ValueTask Seek(ulong publishTime, CancellationToken cancellationToken)
+        {
+            ThrowIfDisposed();
+
+            var seek = new CommandSeek { MessagePublishTime = publishTime };
+            _ = await _executor.Execute(() => _channel.Send(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
+        }
+
+        public async ValueTask Seek(DateTime publishTime, CancellationToken cancellationToken)
+        {
+            ThrowIfDisposed();
+
+            var seek = new CommandSeek { MessagePublishTime = (ulong) new DateTimeOffset(publishTime).ToUnixTimeMilliseconds() };
+            _ = await _executor.Execute(() => _channel.Send(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
+        }
+
+        public async ValueTask Seek(DateTimeOffset publishTime, CancellationToken cancellationToken)
+        {
+            ThrowIfDisposed();
+
+            var seek = new CommandSeek { MessagePublishTime = (ulong) publishTime.ToUnixTimeMilliseconds() };
             _ = await _executor.Execute(() => _channel.Send(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
         }
 
@@ -170,11 +190,7 @@
 
             var redeliverUnacknowledgedMessages = new CommandRedeliverUnacknowledgedMessages();
             redeliverUnacknowledgedMessages.MessageIds.AddRange(messageIds);
-
-            await _executor.Execute(() =>
-            {
-                return _channel.Send(redeliverUnacknowledgedMessages, cancellationToken);
-            }, cancellationToken).ConfigureAwait(false);
+            await _executor.Execute(() => _channel.Send(redeliverUnacknowledgedMessages, cancellationToken), cancellationToken).ConfigureAwait(false);
         }
 
         internal async ValueTask SetChannel(IConsumerChannel channel)
diff --git a/src/DotPulsar/Internal/Reader.cs b/src/DotPulsar/Internal/Reader.cs
index 82f0e91..3dccb73 100644
--- a/src/DotPulsar/Internal/Reader.cs
+++ b/src/DotPulsar/Internal/Reader.cs
@@ -17,6 +17,7 @@
     using Abstractions;
     using DotPulsar.Abstractions;
     using DotPulsar.Exceptions;
+    using DotPulsar.Internal.PulsarApi;
     using Events;
     using System;
     using System.Collections.Generic;
@@ -80,6 +81,38 @@
                 yield return await _executor.Execute(() => _channel.Receive(cancellationToken), cancellationToken).ConfigureAwait(false);
         }
 
+        public async ValueTask Seek(MessageId messageId, CancellationToken cancellationToken)
+        {
+            ThrowIfDisposed();
+
+            var seek = new CommandSeek { MessageId = messageId.Data };
+            _ = await _executor.Execute(() => _channel.Send(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
+        }
+
+        public async ValueTask Seek(ulong publishTime, CancellationToken cancellationToken)
+        {
+            ThrowIfDisposed();
+
+            var seek = new CommandSeek { MessagePublishTime = publishTime };
+            _ = await _executor.Execute(() => _channel.Send(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
+        }
+
+        public async ValueTask Seek(DateTime publishTime, CancellationToken cancellationToken)
+        {
+            ThrowIfDisposed();
+
+            var seek = new CommandSeek { MessagePublishTime = (ulong) new DateTimeOffset(publishTime).ToUnixTimeMilliseconds() };
+            _ = await _executor.Execute(() => _channel.Send(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
+        }
+
+        public async ValueTask Seek(DateTimeOffset publishTime, CancellationToken cancellationToken)
+        {
+            ThrowIfDisposed();
+
+            var seek = new CommandSeek { MessagePublishTime = (ulong) publishTime.ToUnixTimeMilliseconds() };
+            _ = await _executor.Execute(() => _channel.Send(seek, cancellationToken), cancellationToken).ConfigureAwait(false);
+        }
+
         public async ValueTask DisposeAsync()
         {
             if (Interlocked.Exchange(ref _isDisposed, 1) != 0)