diff --git a/core/Processors/IRecordTimestampExtractor.cs b/core/Processors/IRecordTimestampExtractor.cs
new file mode 100644
index 00000000..1cd44dae
--- /dev/null
+++ b/core/Processors/IRecordTimestampExtractor.cs
@@ -0,0 +1,20 @@
+namespace Streamiz.Kafka.Net.Processors
+{
+    /// <summary>
+    /// An interface that allows to dynamically determine the timestamp of the record stored in the Kafka topic.
+    /// </summary>
+    /// <typeparam name="K">Key type</typeparam>
+    /// <typeparam name="V">Value type</typeparam>
+    public interface IRecordTimestampExtractor<K,V>
+    {
+        /// <summary>
+        /// Extracts the timestamp of the record stored in the Kafka topic.
+        /// </summary>
+        /// <param name="key">the record key</param>
+        /// <param name="value">the record value</param>
+        /// <param name="recordContext">current context metadata of the record</param>
+        /// <returns>the timestamp of the record</returns>
+        long Extract(K key, V value, IRecordContext recordContext);
+
+    }
+}
diff --git a/core/Processors/Internal/DefaultRecordTimestampExtractor.cs b/core/Processors/Internal/DefaultRecordTimestampExtractor.cs
new file mode 100644
index 00000000..942b1df3
--- /dev/null
+++ b/core/Processors/Internal/DefaultRecordTimestampExtractor.cs
@@ -0,0 +1,16 @@
+using System;
+
+namespace Streamiz.Kafka.Net.Processors.Internal
+{
+    internal class DefaultRecordTimestampExtractor<K, V> : IRecordTimestampExtractor<K, V>
+    {
+        private readonly Func<K, V, IRecordContext, long> timestampExtractor;
+
+        public DefaultRecordTimestampExtractor()
+        {
+            this.timestampExtractor = (k, v, ctx) => ctx.Timestamp;
+        }
+
+        public long Extract(K key, V value, IRecordContext recordContext) => timestampExtractor(key, value, recordContext);
+    }
+}
diff --git a/core/Processors/Internal/InternalTopologyBuilder.cs b/core/Processors/Internal/InternalTopologyBuilder.cs
index b779204a..e426295a 100644
--- a/core/Processors/Internal/InternalTopologyBuilder.cs
+++ b/core/Processors/Internal/InternalTopologyBuilder.cs
@@ -153,7 +153,7 @@ internal void AddSourceOperator<K, V>(string topic, string nameNode, ConsumedInt
             nodeGroups = null;
         }
 
-        internal void AddSinkOperator<K, V>(ITopicNameExtractor<K, V> topicNameExtractor, string nameNode, Produced<K, V> produced, params string[] previousProcessorNames)
+        internal void AddSinkOperator<K, V>(ITopicNameExtractor<K, V> topicNameExtractor, IRecordTimestampExtractor<K, V> timestampExtractor, string nameNode, Produced<K, V> produced, params string[] previousProcessorNames)
         {
             if (nodeFactories.ContainsKey(nameNode))
             {
@@ -161,7 +161,7 @@ internal void AddSinkOperator<K, V>(ITopicNameExtractor<K, V> topicNameExtractor
             }
 
             nodeFactories.Add(nameNode,
-                new SinkNodeFactory<K, V>(nameNode, previousProcessorNames, topicNameExtractor, produced.KeySerdes, produced.ValueSerdes, produced.Partitioner));
+                new SinkNodeFactory<K, V>(nameNode, previousProcessorNames, topicNameExtractor, timestampExtractor, produced.KeySerdes, produced.ValueSerdes, produced.Partitioner));
             nodeGrouper.Add(nameNode);
             nodeGrouper.Unite(nameNode, previousProcessorNames);
             nodeGroups = null;
diff --git a/core/Processors/Internal/NodeFactory.cs b/core/Processors/Internal/NodeFactory.cs
index f1461736..3527a292 100644
--- a/core/Processors/Internal/NodeFactory.cs
+++ b/core/Processors/Internal/NodeFactory.cs
@@ -76,7 +76,8 @@ internal interface ISinkNodeFactory : INodeFactory
 
     internal class SinkNodeFactory<K, V> : NodeFactory, ISinkNodeFactory
     {
-        public ITopicNameExtractor<K, V> Extractor { get; }
+        public ITopicNameExtractor<K, V> TopicExtractor { get; }
+        public IRecordTimestampExtractor<K, V> TimestampExtractor { get; }
         public ISerDes<K> KeySerdes { get; }
         public ISerDes<V> ValueSerdes { get; }
         public Func<string, K, V, int> ProducedPartitioner { get; }
@@ -85,29 +86,31 @@ public string Topic
         {
             get
             {
-                return Extractor is StaticTopicNameExtractor<K, V> ?
-                    ((StaticTopicNameExtractor<K, V>)Extractor).TopicName :
+                return TopicExtractor is StaticTopicNameExtractor<K, V> ?
+                    ((StaticTopicNameExtractor<K, V>)TopicExtractor).TopicName :
                      null;
             }
         }
 
-        public SinkNodeFactory(string name, string[] previous, ITopicNameExtractor<K, V> topicExtractor,
+        public SinkNodeFactory(string name, string[] previous, ITopicNameExtractor<K, V> topicExtractor, 
+            IRecordTimestampExtractor<K, V> timestampExtractor,
             ISerDes<K> keySerdes, ISerDes<V> valueSerdes, Func<string, K, V, int> producedPartitioner)
             : base(name, previous)
         {
-            Extractor = topicExtractor;
+            TopicExtractor = topicExtractor;
+            TimestampExtractor = timestampExtractor;
             KeySerdes = keySerdes;
             ValueSerdes = valueSerdes;
             ProducedPartitioner = producedPartitioner;
         }
 
         public override IProcessor Build()
-            => new SinkProcessor<K, V>(Name, Extractor, KeySerdes, ValueSerdes, ProducedPartitioner);
+            => new SinkProcessor<K, V>(Name, TopicExtractor, TimestampExtractor, KeySerdes, ValueSerdes, ProducedPartitioner);
 
         public override NodeDescription Describe()
-            => Extractor is StaticTopicNameExtractor<K, V> ?
-            new SinkNodeDescription(Name, ((StaticTopicNameExtractor<K, V>)Extractor).TopicName) :
-            new SinkNodeDescription(Name, Extractor?.GetType());
+            => TopicExtractor is StaticTopicNameExtractor<K, V> ?
+            new SinkNodeDescription(Name, ((StaticTopicNameExtractor<K, V>)TopicExtractor).TopicName) :
+            new SinkNodeDescription(Name, TopicExtractor?.GetType());
     }
 
     #endregion
diff --git a/core/Processors/Internal/WrapperRecordTimestampExtractor.cs b/core/Processors/Internal/WrapperRecordTimestampExtractor.cs
new file mode 100644
index 00000000..39c969c4
--- /dev/null
+++ b/core/Processors/Internal/WrapperRecordTimestampExtractor.cs
@@ -0,0 +1,16 @@
+using System;
+
+namespace Streamiz.Kafka.Net.Processors.Internal
+{
+    internal class WrapperRecordTimestampExtractor<K, V> : IRecordTimestampExtractor<K, V>
+    {
+        private readonly Func<K, V, IRecordContext, long> timestampExtractor;
+
+        public WrapperRecordTimestampExtractor(Func<K,V,IRecordContext,long> timestampExtractor)
+        {
+            this.timestampExtractor = timestampExtractor;
+        }
+
+        public long Extract(K key, V value, IRecordContext recordContext) => timestampExtractor(key, value ,recordContext);
+    }
+}
diff --git a/core/Processors/SinkProcessor.cs b/core/Processors/SinkProcessor.cs
index a7ca8077..63c4a128 100644
--- a/core/Processors/SinkProcessor.cs
+++ b/core/Processors/SinkProcessor.cs
@@ -14,12 +14,14 @@ internal interface ISinkProcessor
     internal class SinkProcessor<K, V> : AbstractProcessor<K, V>, ISinkProcessor
     {
         private ITopicNameExtractor<K, V> topicNameExtractor;
+        private readonly IRecordTimestampExtractor<K, V> timestampExtractor;
         private readonly Func<string, K, V, int> partitioner; 
 
-        internal SinkProcessor(string name, ITopicNameExtractor<K, V> topicNameExtractor, ISerDes<K> keySerdes, ISerDes<V> valueSerdes, Func<string, K, V, int> partitioner = null)
+        internal SinkProcessor(string name, ITopicNameExtractor<K, V> topicNameExtractor, IRecordTimestampExtractor<K, V> timestampExtractor, ISerDes<K> keySerdes, ISerDes<V> valueSerdes, Func<string, K, V, int> partitioner = null)
             : base(name, keySerdes, valueSerdes)
         {
             this.topicNameExtractor = topicNameExtractor;
+            this.timestampExtractor = timestampExtractor;
             this.partitioner = partitioner;
         }
 
@@ -55,8 +57,7 @@ public override void Process(K key, V value)
             }
 
             var topicName = topicNameExtractor.Extract(key, value, Context.RecordContext);
-
-            var timestamp = Context.Timestamp;
+            var timestamp = timestampExtractor.Extract(key, value, Context.RecordContext);
             if (timestamp < 0)
             {
                 throw new StreamsException($"Invalid (negative) timestamp of {timestamp} for output record <{key}:{value}>.");
diff --git a/core/Stream/IKStream.cs b/core/Stream/IKStream.cs
index 52770626..f2972074 100644
--- a/core/Stream/IKStream.cs
+++ b/core/Stream/IKStream.cs
@@ -8,6 +8,7 @@
 using System.Collections.Generic;
 using System.Threading.Tasks;
 using Streamiz.Kafka.Net.Processors.Public;
+using Streamiz.Kafka.Net.Processors.Internal;
 
 namespace Streamiz.Kafka.Net.Stream
 {
@@ -160,8 +161,47 @@ public interface IKStream<K, V>
         /// <param name="keySerdes">Key serializer</param>
         /// <param name="valueSerdes">Value serializer</param>
         /// <param name="named">A <see cref="string"/> config used to name the processor in the topology. Default : null</param>
-        void To(ITopicNameExtractor<K, V> topicExtractor, ISerDes<K> keySerdes, ISerDes<V> valueSerdes, string named = null);
-
+        void To(ITopicNameExtractor<K, V> topicExtractor, ISerDes<K> keySerdes, ISerDes<V> valueSerdes, string named = null);
+
+        /// <summary>
+        /// Dynamically materialize this stream to topics using default serializers specified in the config and producer's.
+        /// The topic names for each record to send to is dynamically determined based on the <code>Func&lt;K, V, IRecordContext, string&gt;</code>.
+        /// </summary>
+        /// <param name="topicExtractor">Extractor function to determine the name of the Kafka topic to write to for each record</param>
+        /// <param name="recordTimestampExtractor">Extractor function to determine the timestamp of the record stored in the Kafka topic</param>
+        /// <param name="named">A <see cref="string"/> config used to name the processor in the topology. Default : null</param>
+        void To(Func<K, V, IRecordContext, string> topicExtractor, Func<K, V, IRecordContext, long> recordTimestampExtractor, string named = null);
+
+        /// <summary>
+        /// Dynamically materialize this stream to topics using default serializers specified in the config and producer's.
+        /// The topic names for each record to send to is dynamically determined based on the <code>Func&lt;K, V, IRecordContext, string&gt;</code>.
+        /// </summary>
+        /// <param name="topicExtractor">Extractor function to determine the name of the Kafka topic to write to for each record</param>
+        /// <param name="keySerdes">Key serializer</param>
+        /// <param name="valueSerdes">Value serializer</param>
+        /// <param name="recordTimestampExtractor">Extractor function to determine the timestamp of the record stored in the Kafka topic</param>
+        /// <param name="named">A <see cref="string"/> config used to name the processor in the topology. Default : null</param>
+        void To(Func<K, V, IRecordContext, string> topicExtractor, ISerDes<K> keySerdes, ISerDes<V> valueSerdes, Func<K, V, IRecordContext, long> recordTimestampExtractor, string named = null);
+
+        /// <summary>
+        /// Dynamically materialize this stream to topics using default serializers specified in the config and producer's.
+        /// The topic names for each record to send to is dynamically determined based on the <see cref="ITopicNameExtractor&lt;K, V&gt;"/>}.
+        /// </summary>
+        /// <param name="topicExtractor">The extractor to determine the name of the Kafka topic to write to for each record</param>
+        /// <param name="recordTimestampExtractor">Extractor function to determine the timestamp of the record stored in the Kafka topic</param>
+        /// <param name="named">A <see cref="string"/> config used to name the processor in the topology. Default : null</param>
+        void To(ITopicNameExtractor<K, V> topicExtractor, IRecordTimestampExtractor<K, V> recordTimestampExtractor, string named = null);
+
+        /// <summary>
+        /// Dynamically materialize this stream to a topic using serializers specified in the method parameters.
+        /// The topic names for each record to send to is dynamically determined based on the <see cref="ITopicNameExtractor&lt;K, V&gt;"/>}.
+        /// </summary>
+        /// <param name="topicExtractor">The extractor to determine the name of the Kafka topic to write to for each record</param>4
+        /// <param name="keySerdes">Key serializer</param>
+        /// <param name="valueSerdes">Value serializer</param>
+        /// <param name="recordTimestampExtractor">Extractor function to determine the timestamp of the record stored in the Kafka topic</param>
+        /// <param name="named">A <see cref="string"/> config used to name the processor in the topology. Default : null</param>
+        void To(ITopicNameExtractor<K, V> topicExtractor, ISerDes<K> keySerdes, ISerDes<V> valueSerdes, IRecordTimestampExtractor<K, V> recordTimestampExtractor, string named = null);
 
         /// <summary>
         /// Materialize this stream to a topic using <typeparamref name="KS"/> and <typeparamref name="VS"/> serializers specified in the method parameters.
@@ -192,12 +232,34 @@ public interface IKStream<K, V>
         /// <typeparam name="VS">New type value serializer</typeparam>
         /// <param name="topicExtractor">The extractor to determine the name of the Kafka topic to write to for each record</param>
         /// <param name="named">A <see cref="string"/> config used to name the processor in the topology. Default : null</param>
-        void To<KS, VS>(ITopicNameExtractor<K, V> topicExtractor, string named = null) where KS : ISerDes<K>, new() where VS : ISerDes<V>, new();
-
-        #endregion
+        void To<KS, VS>(ITopicNameExtractor<K, V> topicExtractor, string named = null) where KS : ISerDes<K>, new() where VS : ISerDes<V>, new();
+
+        /// <summary>
+        /// Dynamically materialize this stream to a topic using <typeparamref name="KS"/> and <typeparamref name="VS"/> serializers specified in the method parameters.
+        /// The topic names for each record to send to is dynamically determined based on the <code>Func&lt;K, V, IRecordContext, string&gt;</code>.
+        /// </summary>
+        /// <typeparam name="KS">New type key serializer</typeparam>
+        /// <typeparam name="VS">New type value serializer</typeparam>
+        /// <param name="topicExtractor">Extractor function to determine the name of the Kafka topic to write to for each record</param>
+        /// <param name="recordTimestampExtractor">Extractor function to determine the timestamp of the record stored in the Kafka topic</param>
+        /// <param name="named">A <see cref="string"/> config used to name the processor in the topology. Default : null</param>
+        void To<KS, VS>(Func<K, V, IRecordContext, string> topicExtractor, Func<K, V, IRecordContext, long> recordTimestampExtractor, string named = null) where KS : ISerDes<K>, new() where VS : ISerDes<V>, new();
+
+        /// <summary>
+        /// Dynamically materialize this stream to a topic using <typeparamref name="KS"/> and <typeparamref name="VS"/> serializers specified in the method parameters.
+        /// The topic names for each record to send to is dynamically determined based on the <see cref="ITopicNameExtractor&lt;K, V&gt;"/>}.
+        /// </summary>
+        /// <typeparam name="KS">New type key serializer</typeparam>
+        /// <typeparam name="VS">New type value serializer</typeparam>
+        /// <param name="topicExtractor">The extractor to determine the name of the Kafka topic to write to for each record</param>
+        /// <param name="recordTimestampExtractor">Extractor function to determine the timestamp of the record stored in the Kafka topic</param>
+        /// <param name="named">A <see cref="string"/> config used to name the processor in the topology. Default : null</param>
+        void To<KS, VS>(ITopicNameExtractor<K, V> topicExtractor, IRecordTimestampExtractor<K, V> recordTimestampExtractor, string named = null) where KS : ISerDes<K>, new() where VS : ISerDes<V>, new();
         
+        #endregion
+
         #region FlatMap
-
+
         /// <summary>
         /// Transform each record of the input stream into zero or more records in the output stream (bot
         /// can be altered arbitrarily).
diff --git a/core/Stream/Internal/Graph/Nodes/AsyncNode.cs b/core/Stream/Internal/Graph/Nodes/AsyncNode.cs
index ce894a84..b58e5a13 100644
--- a/core/Stream/Internal/Graph/Nodes/AsyncNode.cs
+++ b/core/Stream/Internal/Graph/Nodes/AsyncNode.cs
@@ -34,6 +34,7 @@ public override void WriteToTopology(InternalTopologyBuilder builder)
                 builder.AddInternalTopic(RepartitionTopic, null);
                 builder.AddSinkOperator(
                     new StaticTopicNameExtractor<TK, TV>(RepartitionTopic),
+                    new DefaultRecordTimestampExtractor<TK, TV>(),
                     SinkName,
                     Produced<TK, TV>.Create(KeySerdes, ValueSerdes),
                     ParentNodeNames());
@@ -70,6 +71,7 @@ public override void WriteToTopology(InternalTopologyBuilder builder)
                 builder.AddInternalTopic(RepartitionTopic, null);
                 builder.AddProcessor(ProcessorParameters.ProcessorName, ProcessorParameters.Processor, ParentNodeNames());
                 builder.AddSinkOperator(new StaticTopicNameExtractor<TK1, TV1>(RepartitionTopic),
+                    new DefaultRecordTimestampExtractor<TK1, TV1>(),
                         SinkName,
                         Produced<TK1, TV1>.Create(KeySerdes, ValueSerdes),
                         ProcessorParameters.ProcessorName);
@@ -112,6 +114,7 @@ public override void WriteToTopology(InternalTopologyBuilder builder)
             {
                 builder.AddInternalTopic(RequestTopic, null);
                 builder.AddSinkOperator(new StaticTopicNameExtractor<TK, TV>(RequestTopic),
+                    new DefaultRecordTimestampExtractor<TK, TV>(),
                     SinkName,
                     Produced<TK, TV>.Create(KeySerdes, ValueSerdes),
                     ParentNodeNames());
diff --git a/core/Stream/Internal/Graph/Nodes/RepartitionNode.cs b/core/Stream/Internal/Graph/Nodes/RepartitionNode.cs
index 8c87d51d..ce4f14e9 100644
--- a/core/Stream/Internal/Graph/Nodes/RepartitionNode.cs
+++ b/core/Stream/Internal/Graph/Nodes/RepartitionNode.cs
@@ -23,6 +23,7 @@ public override void WriteToTopology(InternalTopologyBuilder builder)
                     ParentNodeNames());
                 builder.AddSinkOperator(
                     new StaticTopicNameExtractor<K, V>(RepartitionTopic),
+                    new DefaultRecordTimestampExtractor<K, V>(),
                     SinkName,
                     Produced<K, V>.Create(KeySerdes, ValueSerdes).WithPartitioner(StreamPartitioner),
                     ProcessorParameters.ProcessorName);
@@ -36,6 +37,7 @@ public override void WriteToTopology(InternalTopologyBuilder builder)
                 builder.AddInternalTopic(RepartitionTopic, NumberOfPartition);
                 builder.AddSinkOperator(
                     new StaticTopicNameExtractor<K, V>(RepartitionTopic),
+                    new DefaultRecordTimestampExtractor<K, V>(),
                     SinkName,
                     Produced<K, V>.Create(KeySerdes, ValueSerdes).WithPartitioner(StreamPartitioner),
                     ParentNodeNames());
diff --git a/core/Stream/Internal/Graph/Nodes/StreamSinkNode.cs b/core/Stream/Internal/Graph/Nodes/StreamSinkNode.cs
index 743892a9..2ed39ac2 100644
--- a/core/Stream/Internal/Graph/Nodes/StreamSinkNode.cs
+++ b/core/Stream/Internal/Graph/Nodes/StreamSinkNode.cs
@@ -15,17 +15,19 @@ internal class StreamSinkNode<K, V> : StreamSinkNode
     {
         private readonly ITopicNameExtractor<K, V> topicNameExtractor;
         private readonly Produced<K, V> produced;
+        private readonly IRecordTimestampExtractor<K, V> timestampExtractor;
 
-        public StreamSinkNode(ITopicNameExtractor<K, V> topicNameExtractor, string streamGraphNode, Produced<K, V> produced)
+        public StreamSinkNode(ITopicNameExtractor<K, V> topicNameExtractor, IRecordTimestampExtractor<K, V> timestampExtractor, string streamGraphNode, Produced<K, V> produced)
             : base(streamGraphNode)
         {
             this.topicNameExtractor = topicNameExtractor;
+            this.timestampExtractor = timestampExtractor;
             this.produced = produced;
         }
 
         public override void WriteToTopology(InternalTopologyBuilder builder)
         {
-            builder.AddSinkOperator(topicNameExtractor, this.streamGraphNode, produced, ParentNodeNames());
+            builder.AddSinkOperator(topicNameExtractor, timestampExtractor, this.streamGraphNode, produced, ParentNodeNames());
         }
     }
 }
diff --git a/core/Stream/Internal/KStream.cs b/core/Stream/Internal/KStream.cs
index 19f7642d..b3bb2076 100644
--- a/core/Stream/Internal/KStream.cs
+++ b/core/Stream/Internal/KStream.cs
@@ -206,15 +206,27 @@ public void To(string topicName, ISerDes<K> keySerdes, ISerDes<V> valueSerdes, s
             To(new StaticTopicNameExtractor<K, V>(topicName), keySerdes, valueSerdes, named);
         }
 
-        public void To(ITopicNameExtractor<K, V> topicExtractor, string named = null) => DoTo(topicExtractor, Produced<K, V>.Create(KeySerdes, ValueSerdes).WithName(named));
+        public void To(ITopicNameExtractor<K, V> topicExtractor, string named = null) => DoTo(topicExtractor, new DefaultRecordTimestampExtractor<K, V>(), Produced<K, V>.Create(KeySerdes, ValueSerdes).WithName(named));
 
         public void To(ITopicNameExtractor<K, V> topicExtractor, ISerDes<K> keySerdes, ISerDes<V> valueSerdes, string named = null)
-            => DoTo(topicExtractor, Produced<K,V>.Create(keySerdes, valueSerdes).WithName(named));
+            => DoTo(topicExtractor, new DefaultRecordTimestampExtractor<K, V>(), Produced<K,V>.Create(keySerdes, valueSerdes).WithName(named));
 
         public void To(Func<K, V, IRecordContext, string> topicExtractor, string named = null) => To(new WrapperTopicNameExtractor<K, V>(topicExtractor), named);
 
         public void To(Func<K, V, IRecordContext, string> topicExtractor, ISerDes<K> keySerdes, ISerDes<V> valueSerdes, string named = null)
-            => To(new WrapperTopicNameExtractor<K, V>(topicExtractor), keySerdes, valueSerdes, named);
+            => To(new WrapperTopicNameExtractor<K, V>(topicExtractor), keySerdes, valueSerdes, named);
+
+        public void To(Func<K, V, IRecordContext, string> topicExtractor, Func<K, V, IRecordContext, long> recordTimestampExtractor, string named = null) =>
+            DoTo(new WrapperTopicNameExtractor<K, V>(topicExtractor), new WrapperRecordTimestampExtractor<K, V>(recordTimestampExtractor), Produced<K, V>.Create(KeySerdes, ValueSerdes).WithName(named));
+
+        public void To(Func<K, V, IRecordContext, string> topicExtractor, ISerDes<K> keySerdes, ISerDes<V> valueSerdes, Func<K, V, IRecordContext, long> recordTimestampExtractor, string named = null) =>
+            DoTo(new WrapperTopicNameExtractor<K, V>(topicExtractor), new WrapperRecordTimestampExtractor<K, V>(recordTimestampExtractor), Produced<K, V>.Create(keySerdes, valueSerdes).WithName(named));
+
+        public void To(ITopicNameExtractor<K, V> topicExtractor, IRecordTimestampExtractor<K, V> recordTimestampExtractor, string named = null) =>
+            DoTo(topicExtractor, recordTimestampExtractor, Produced<K, V>.Create(KeySerdes, ValueSerdes).WithName(named));
+
+        public void To(ITopicNameExtractor<K, V> topicExtractor, ISerDes<K> keySerdes, ISerDes<V> valueSerdes, IRecordTimestampExtractor<K, V> recordTimestampExtractor, string named = null) =>
+            DoTo(topicExtractor, recordTimestampExtractor, Produced<K, V>.Create(keySerdes, valueSerdes).WithName(named));    
 
         public void To<KS, VS>(Func<K, V, IRecordContext, string> topicExtractor, string named = null)
             where KS : ISerDes<K>, new()
@@ -229,7 +241,17 @@ public void To<KS, VS>(string topicName, string named = null)
         public void To<KS, VS>(ITopicNameExtractor<K, V> topicExtractor, string named = null)
             where KS : ISerDes<K>, new()
             where VS : ISerDes<V>, new()
-            => DoTo(topicExtractor, Produced<K, V>.Create<KS, VS>().WithName(named));
+            => DoTo(topicExtractor, new DefaultRecordTimestampExtractor<K, V>(), Produced<K, V>.Create<KS, VS>().WithName(named));
+
+        public void To<KS, VS>(Func<K, V, IRecordContext, string> topicExtractor, Func<K, V, IRecordContext, long> recordTimestampExtractor, string named = null)
+            where KS : ISerDes<K>, new()
+            where VS : ISerDes<V>, new()
+            => To<KS, VS>(new WrapperTopicNameExtractor<K, V>(topicExtractor), new WrapperRecordTimestampExtractor<K, V>(recordTimestampExtractor), named);
+
+        public void To<KS, VS>(ITopicNameExtractor<K, V> topicExtractor, IRecordTimestampExtractor<K, V> recordTimestampExtractor, string named = null)
+            where KS : ISerDes<K>, new()
+            where VS : ISerDes<V>, new()
+        => DoTo(topicExtractor, recordTimestampExtractor, Produced<K, V>.Create<KS, VS>().WithName(named));
 
         #endregion
 
@@ -1036,13 +1058,13 @@ private IKStream<K, V> DoFilter(Func<K, V, bool> predicate, string named, bool n
 
             builder.AddGraphNode(Node, filterProcessorNode);
             return new KStream<K, V>(name, KeySerdes, ValueSerdes, SetSourceNodes, RepartitionRequired, filterProcessorNode, builder);
-        }
-
-        private void DoTo(ITopicNameExtractor<K, V> topicExtractor, Produced<K, V> produced)
+        }
+
+        private void DoTo(ITopicNameExtractor<K, V> topicExtractor, IRecordTimestampExtractor<K, V> timestampExtractor, Produced<K, V> produced)
         {
             string name = new Named(produced.Named).OrElseGenerateWithPrefix(builder, KStream.SINK_NAME);
 
-            StreamSinkNode<K, V> sinkNode = new StreamSinkNode<K, V>(topicExtractor, name, produced);
+            StreamSinkNode<K, V> sinkNode = new StreamSinkNode<K, V>(topicExtractor, timestampExtractor, name, produced);
             builder.AddGraphNode(Node, sinkNode);
         }
 
@@ -1320,8 +1342,8 @@ string asyncProcessorName
                 responseSinkProcessorName,
                 responseSourceProcessorName,
                 asyncProcessorName);
-        }
-
+        }
+
         #endregion
     }
 }
\ No newline at end of file
diff --git a/test/Streamiz.Kafka.Net.Tests/Private/StreamGraphNodeTests.cs b/test/Streamiz.Kafka.Net.Tests/Private/StreamGraphNodeTests.cs
index e099bdab..4a8dc36d 100644
--- a/test/Streamiz.Kafka.Net.Tests/Private/StreamGraphNodeTests.cs
+++ b/test/Streamiz.Kafka.Net.Tests/Private/StreamGraphNodeTests.cs
@@ -77,7 +77,9 @@ public void WriteTopologyTest()
             nodes.Add(filter);
 
             var to = new StreamSinkNode<string, string>(
-                new StaticTopicNameExtractor<string, string>("topic2"), "to-03",
+                new StaticTopicNameExtractor<string, string>("topic2"),
+                new DefaultRecordTimestampExtractor<string, string>(),
+                "to-03",
                 new Stream.Internal.Produced<string, string>(
                     new StringSerDes(),
                     new StringSerDes())
diff --git a/test/Streamiz.Kafka.Net.Tests/Processors/KStreamRecordTimestampExtractorTests.cs b/test/Streamiz.Kafka.Net.Tests/Processors/KStreamRecordTimestampExtractorTests.cs
new file mode 100644
index 00000000..2c4cbed4
--- /dev/null
+++ b/test/Streamiz.Kafka.Net.Tests/Processors/KStreamRecordTimestampExtractorTests.cs
@@ -0,0 +1,120 @@
+using Confluent.Kafka;
+using NUnit.Framework;
+using Streamiz.Kafka.Net.Mock;
+using Streamiz.Kafka.Net.Processors;
+using Streamiz.Kafka.Net.SerDes;
+using Streamiz.Kafka.Net.Stream;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+
+namespace Streamiz.Kafka.Net.Tests.Processors
+{
+    public class KStreamRecordTimestampExtractorTests
+    {
+        private class MyTimestampExtractor : ITimestampExtractor
+        {
+            public long Extract(ConsumeResult<object, object> record, long partitionTime)
+            {
+                return Timestamp.DateTimeToUnixTimestampMs(DateTime.UtcNow);
+            }
+        }
+
+        [TestCase(0)]
+        [TestCase(12345)]
+        [TestCase(int.MaxValue)]
+        public void StreamWithIngestionTimeTest(long customTimestamp)
+        {
+            var builder = new StreamBuilder();
+            var data = new List<KeyValuePair<string, string>>() { KeyValuePair.Create("key1", "123456") };
+
+            builder.Stream<string, string>("topic")
+                .To((k,v,ctx) => "output-topic", (k,v,ctx) => customTimestamp);
+
+            var config = new StreamConfig<StringSerDes, StringSerDes>();
+            config.ApplicationId = "test-flatmap";
+
+            Topology t = builder.Build();
+
+            using (var driver = new TopologyTestDriver(t, config))
+            {
+                var inputTopic = driver.CreateInputTopic<string, string>("topic");
+                var outputTopic = driver.CreateOuputTopic<string, string, StringSerDes, StringSerDes>("output-topic");
+
+                inputTopic.PipeInputs(data);
+                var result = outputTopic.ReadKeyValueList().ToList();
+
+                Assert.IsNotNull(result);
+                Assert.IsTrue(result.Count == 1);
+                var record = result[0];
+                Assert.AreEqual(customTimestamp, record.Message.Timestamp.UnixTimestampMs);
+            }
+        }
+
+        [TestCase(0)]
+        [TestCase(12345)]
+        [TestCase(int.MaxValue)]
+        public void StreamWithStreamTimestampExtractorTest(long customTimestamp)
+        {
+            var builder = new StreamBuilder();
+            var data = new List<KeyValuePair<string, string>>() { KeyValuePair.Create("key1", "123456") };
+
+            builder.Stream<string, string>("topic",
+                new StringSerDes(),
+                new StringSerDes(),
+                new MyTimestampExtractor())
+                .To((k, v, ctx) => "output-topic", (k, v, ctx) => customTimestamp);
+
+            var config = new StreamConfig<StringSerDes, StringSerDes>();
+            config.ApplicationId = "test-flatmap";
+
+            Topology t = builder.Build();
+
+            using (var driver = new TopologyTestDriver(t, config))
+            {
+                var inputTopic = driver.CreateInputTopic<string, string>("topic");
+                var outputTopic = driver.CreateOuputTopic<string, string, StringSerDes, StringSerDes>("output-topic");
+
+                inputTopic.PipeInputs(data);
+                var result = outputTopic.ReadKeyValueList().ToList();
+
+                Assert.IsNotNull(result);
+                Assert.IsTrue(result.Count == 1);
+                var record = result[0];
+                Assert.AreEqual(customTimestamp, record.Message.Timestamp.UnixTimestampMs);
+            }
+        }
+
+        [TestCase(0)]
+        [TestCase(12345)]
+        [TestCase(int.MaxValue)]
+        public void WhenStreamWithOverriddenTimestamp(long customTimestamp)
+        {
+            var builder = new StreamBuilder();
+            var data = new List<KeyValuePair<string, string>>() { KeyValuePair.Create("key1", "123456") };
+
+            builder.Stream<string, string>("topic")
+                .WithRecordTimestamp((k, v) => customTimestamp + 3600)
+                .To((k, v, ctx) => "output-topic", (k, v, ctx) => customTimestamp);
+
+            var config = new StreamConfig<StringSerDes, StringSerDes>();
+            config.ApplicationId = "test-flatmap";
+
+            Topology t = builder.Build();
+
+            using (var driver = new TopologyTestDriver(t, config))
+            {
+                var inputTopic = driver.CreateInputTopic<string, string>("topic");
+                var outputTopic = driver.CreateOuputTopic<string, string, StringSerDes, StringSerDes>("output-topic");
+
+                inputTopic.PipeInputs(data);
+                var result = outputTopic.ReadKeyValueList().ToList();
+
+                Assert.IsNotNull(result);
+                Assert.IsTrue(result.Count == 1);
+                var record = result[0];
+                Assert.AreEqual(customTimestamp, record.Message.Timestamp.UnixTimestampMs);
+            }
+        }
+    }
+}