Add OffsetsForTimes API to Consumer

src/Confluent.Kafka/Consumer.cs

@@ -611,6 +611,31 @@ public WatermarkOffsets QueryWatermarkOffsets(TopicPartition topicPartition, Tim
         public WatermarkOffsets QueryWatermarkOffsets(TopicPartition topicPartition)
             => consumer.QueryWatermarkOffsets(topicPartition);
 
+        /// <summary>
+        ///     Look up the offsets for the given partitions by timestamp. The returned offset for each partition is the
+        ///     earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition.
+        ///
+        /// </summary>
+        /// <remarks>
+        ///     This is a blocking call.The consumer does not have to be assigned the partitions.
+        ///     If the message format version in a partition is before 0.10.0, i.e.the messages do not have timestamps, null
+        ///     will be returned for that partition.
+        ///
+        ///     Notice that this method may block indefinitely if the partition does not exist.
+        /// </remarks>
+        /// <param name="timestampsToSearch">
+        ///     The mapping from partition to the timestamp to look up.
+        /// </param>
+        /// <param name="timeout">
+        ///     The maximum period of time the call may block.
+        /// </param>
+        /// <returns>
+        ///     A mapping from partition to the timestamp and offset of the first message with timestamp greater
+        ///     than or equal to the target timestamp. null will be returned for the partition if there is no such message.
+        /// </returns>
+        public IEnumerable<TopicPartitionOffset> OffsetsForTimes(IEnumerable<TopicPartitionTimestamp> timestampsToSearch, TimeSpan timeout)
+            => consumer.OffsetsForTimes(timestampsToSearch, timeout);
+
         /// <summary>
         ///     Refer to <see cref="Confluent.Kafka.Producer.GetMetadata(bool,string,int)" /> for more information.
         ///     
@@ -1240,6 +1265,31 @@ public GroupInfo ListGroup(string group)
         public WatermarkOffsets GetWatermarkOffsets(TopicPartition topicPartition)
             => kafkaHandle.GetWatermarkOffsets(topicPartition.Topic, topicPartition.Partition);
 
+        /// <summary>
+        ///     Look up the offsets for the given partitions by timestamp. The returned offset for each partition is the
+        ///     earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition.
+        ///
+        /// </summary>
+        /// <remarks>
+        ///     This is a blocking call.The consumer does not have to be assigned the partitions.
+        ///     If the message format version in a partition is before 0.10.0, i.e.the messages do not have timestamps, null
+        ///     will be returned for that partition.
+        ///
+        ///     Notice that this method may block indefinitely if the partition does not exist.
+        /// </remarks>
+        /// <param name="timestampsToSearch">
+        ///     The mapping from partition to the timestamp to look up.
+        /// </param>
+        /// <param name="timeout">
+        ///     The maximum period of time the call may block.
+        /// </param>
+        /// <returns>
+        ///     A mapping from partition to the timestamp and offset of the first message with timestamp greater
+        ///     than or equal to the target timestamp. null will be returned for the partition if there is no such message.
+        /// </returns>
+        public IEnumerable<TopicPartitionOffset> OffsetsForTimes(IEnumerable<TopicPartitionTimestamp> timestampsToSearch, TimeSpan timeout)
+            => kafkaHandle.OffsetsForTimes(timestampsToSearch, timeout.TotalMillisecondsAsInt());
+
         /// <summary>
         ///     Query the Kafka cluster for low (oldest/beginning) and high (newest/end)
         ///     offsets for the specified topic/partition (blocking)

src/Confluent.Kafka/Impl/LibRdKafka.cs

@@ -148,6 +148,7 @@ static LibRdKafka()
             _poll_set_consumer = NativeMethods.rd_kafka_poll_set_consumer;
             _query_watermark_offsets = NativeMethods.rd_kafka_query_watermark_offsets;
             _get_watermark_offsets = NativeMethods.rd_kafka_get_watermark_offsets;
+            _offsets_for_times = NativeMethods.rd_kafka_offsets_for_times;
             _mem_free = NativeMethods.rd_kafka_mem_free;
             _subscribe = NativeMethods.rd_kafka_subscribe;
             _unsubscribe = NativeMethods.rd_kafka_unsubscribe;
@@ -391,6 +392,11 @@ internal static SafeTopicHandle topic_new(IntPtr rk, string topic, IntPtr conf)
                 out long low, out long high)
             => _get_watermark_offsets(rk, topic, partition, out low, out high);
 
+        private delegate ErrorCode OffsetsForTimes(IntPtr rk, IntPtr topics, IntPtr timeout_ms);
+        private static OffsetsForTimes _offsets_for_times;
+        internal static ErrorCode offsets_for_times(IntPtr rk, IntPtr topics, IntPtr timeout_ms)
+            => _offsets_for_times(rk, topics, timeout_ms);
+
         private static Action<IntPtr, IntPtr> _mem_free;
         internal static void mem_free(IntPtr rk, IntPtr ptr)
             => _mem_free(rk, ptr);
@@ -712,6 +718,11 @@ private class NativeMethods
                     [MarshalAs(UnmanagedType.LPStr)] string topic,
                     int partition, out long low, out long high);
 
+            [DllImport(DllName, CallingConvention = CallingConvention.Cdecl)]
+            internal static extern ErrorCode rd_kafka_offsets_for_times(IntPtr rk,
+                /* rd_kafka_topic_partition_list_t * */ IntPtr topics,
+                IntPtr timeout_ms);
+
             [DllImport(DllName, CallingConvention = CallingConvention.Cdecl)]
             internal static extern void rd_kafka_mem_free(IntPtr rk, IntPtr ptr);
 

src/Confluent.Kafka/Impl/SafeKafkaHandle.cs

@@ -258,6 +258,22 @@ internal WatermarkOffsets GetWatermarkOffsets(string topic, int partition)
             return new WatermarkOffsets(low, high);
         }
 
+        internal IEnumerable<TopicPartitionOffset> OffsetsForTimes(IEnumerable<TopicPartitionTimestamp> timestampsToSearch, int millisecondsTimeout)
+        {
+            IEnumerable<TopicPartitionOffset> offsets = timestampsToSearch
+                .Select(t => new TopicPartitionOffset(t.TopicPartition, t.Timestamp.UnixTimestampMs))
+                .ToList();
+            IntPtr cOffsets = GetCTopicPartitionList(offsets);
+            ErrorCode err = LibRdKafka.offsets_for_times(handle, cOffsets, (IntPtr)millisecondsTimeout);
+            if (err != ErrorCode.NoError)
+            {
+                throw new KafkaException(err);
+            }
+
+            var list = GetTopicPartitionOffsetErrorList(cOffsets);
+            return list.Select(t => t.TopicPartitionOffset).ToList();
+        }
+
         internal void Subscribe(IEnumerable<string> topics)
         {
             IntPtr list = LibRdKafka.topic_partition_list_new((IntPtr) topics.Count());

src/Confluent.Kafka/Producer.cs

@@ -1005,8 +1005,12 @@ public class Producer<TKey, TValue> : ISerializingProducer<TKey, TValue>, IDispo
             ISerializer<TValue> valueSerializer,
             bool manualPoll, bool disableDeliveryReports)
         {
-            var configWithoutKeySerializerProperties = KeySerializer.Configure(config, true);
-            var configWithoutValueSerializerProperties = ValueSerializer.Configure(config, false);
+            var configWithoutKeySerializerProperties = keySerializer != null
+                ? keySerializer.Configure(config, true)
+                : Enumerable.Empty<KeyValuePair<string, object>>();
+            var configWithoutValueSerializerProperties = valueSerializer != null
+                ? valueSerializer.Configure(config, false)
+                : Enumerable.Empty<KeyValuePair<string, object>>();
 
             var configWithoutSerializerProperties = config.Where(item => 
                 configWithoutKeySerializerProperties.Any(ci => ci.Key == item.Key) &&

src/Confluent.Kafka/TopicPartitionTimestamp.cs

@@ -0,0 +1,149 @@
+// Copyright 2016-2017 Confluent Inc., 2015-2016 Andreas Heider
+//
+// Licensed 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
+//
+// http://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.
+//
+// Derived from: rdkafka-dotnet, licensed under the 2-clause BSD License.
+//
+// Refer to LICENSE for more information.
+
+namespace Confluent.Kafka
+{
+    /// <summary>
+    ///     Represents a Kafka (topic, partition, timestamp) tuple.
+    /// </summary>
+    public class TopicPartitionTimestamp
+    {
+        /// <summary>
+        ///     Initializes a new TopicPartitionTimestamp instance.
+        /// </summary>
+        /// <param name="tp">
+        ///     Kafka topic name and partition.
+        /// </param>
+        /// <param name="timestamp">
+        ///     A Kafka timestamp value.
+        /// </param>
+        public TopicPartitionTimestamp(TopicPartition tp, Timestamp timestamp)
+            : this (tp.Topic, tp.Partition, timestamp)
+        {
+        }
+
+        /// <summary>
+        ///     Initializes a new TopicPartitionTimestamp instance.
+        /// </summary>
+        /// <param name="topic">
+        ///     A Kafka topic name.
+        /// </param>
+        /// <param name="partition">
+        ///     A Kafka partition.
+        /// </param>
+        /// <param name="timestamp">
+        ///     A Kafka timestamp value.
+        /// </param>
+        public TopicPartitionTimestamp(string topic, int partition, Timestamp timestamp)
+        {
+            Topic = topic;
+            Partition = partition;
+            Timestamp = timestamp;
+        }
+
+        /// <summary>
+        ///     Gets the Kafka topic name.
+        /// </summary>
+        public string Topic { get; }
+
+        /// <summary>
+        ///     Gets the Kafka partition.
+        /// </summary>
+        public int Partition { get; }
+
+        /// <summary>
+        ///     Gets the Kafka timestamp.
+        /// </summary>
+        public Timestamp Timestamp { get; }
+
+        /// <summary>
+        ///     Gets the TopicPartition component of this TopicPartitionTimestamp instance.
+        /// </summary>
+        public TopicPartition TopicPartition
+            => new TopicPartition(Topic, Partition);
+
+        /// <summary>
+        ///     Tests whether this TopicPartitionTimestamp instance is equal to the specified object.
+        /// </summary>
+        /// <param name="obj">
+        ///     The object to test.
+        /// </param>
+        /// <returns>
+        ///     true if obj is a TopicPartitionTimestamp and all properties are equal. false otherwise.
+        /// </returns>
+        public override bool Equals(object obj)
+        {
+            if (!(obj is TopicPartitionTimestamp))
+            {
+                return false;
+            }
+
+            var tp = (TopicPartitionTimestamp)obj;
+            return tp.Partition == Partition && tp.Topic == Topic && tp.Timestamp == Timestamp;
+        }
+
+        /// <summary>
+        ///     Returns a hash code for this TopicPartitionTimestamp.
+        /// </summary>
+        /// <returns>
+        ///     An integer that specifies a hash value for this TopicPartitionTimestamp.
+        /// </returns>
+        public override int GetHashCode()
+            // x by prime number is quick and gives decent distribution.
+            => (Partition.GetHashCode() * 251 + Topic.GetHashCode()) * 251 + Timestamp.GetHashCode();
+
+        /// <summary>
+        ///     Tests whether TopicPartitionTimestamp instance a is equal to TopicPartitionTimestamp instance b.
+        /// </summary>
+        /// <param name="a">
+        ///     The first TopicPartitionTimestamp instance to compare.
+        /// </param>
+        /// <param name="b">
+        ///     The second TopicPartitionTimestamp instance to compare.
+        /// </param>
+        /// <returns>
+        ///     true if TopicPartitionTimestamp instances a and b are equal. false otherwise.
+        /// </returns>
+        public static bool operator ==(TopicPartitionTimestamp a, TopicPartitionTimestamp b)
+            => a.Equals(b);
+
+        /// <summary>
+        ///     Tests whether TopicPartitionTimestamp instance a is not equal to TopicPartitionTimestamp instance b.
+        /// </summary>
+        /// <param name="a">
+        ///     The first TopicPartitionTimestamp instance to compare.
+        /// </param>
+        /// <param name="b">
+        ///     The second TopicPartitionTimestamp instance to compare.
+        /// </param>
+        /// <returns>
+        ///     true if TopicPartitionTimestamp instances a and b are not equal. false otherwise.
+        /// </returns>
+        public static bool operator !=(TopicPartitionTimestamp a, TopicPartitionTimestamp b)
+            => !(a == b);
+
+        /// <summary>
+        ///     Returns a string representation of the TopicPartitionTimestamp object.
+        /// </summary>
+        /// <returns>
+        ///     A string that represents the TopicPartitionTimestamp object.
+        /// </returns>
+        public override string ToString()
+            => $"{Topic} [{Partition}] @{Timestamp}";
+    }
+}