Fix IConsumer breaking change (#2071)

* fix: remove breaking change to IConsumer

A method was added to the IConsumer interface by #2027 that resulted in
a breaking change in the 2.1 release. This replaces the interface
method and the implementation on Consumer with an extension method on
IConsumer to ensure that existing implementations of IConsumer remain
unbroken and have the same functionality as the default Consumer
implementation.

* Add CHANGELOG

* Remove unused import

---------

Co-authored-by: ttd2089 <[email protected]>

Author: anchitj

CHANGELOG.md

@@ -12,6 +12,7 @@
 ## Fixes
 
 - Fix backwards compatability of TopicPartitionOffset constructor. ([drinehimer](https://github.com/drinehimer), #2066)
+- Fix IConsumer breaking change. ([ttd2089](https://github.com/ttd2089), #2071)
 
 
 # 2.1.1

src/Confluent.Kafka/Consumer.cs

@@ -263,7 +263,7 @@ private void RebalanceCallback(
                     {
                         try
                         {
-                            assignmentWithPositions.Add(PositionTopicPartitionOffset(tp));
+                            assignmentWithPositions.Add(this.PositionTopicPartitionOffset(tp));
                         }
                         catch
                         {
@@ -509,24 +509,17 @@ public List<TopicPartitionOffset> Committed(IEnumerable<TopicPartition> partitio
 
         /// <inheritdoc/>
         public Offset Position(TopicPartition partition)
-        {
-            return PositionTopicPartitionOffset(partition).Offset;
-        }
-        
-        /// <inheritdoc/>
-        public TopicPartitionOffset PositionTopicPartitionOffset(TopicPartition partition)
         {
             try
             {
-                return kafkaHandle.Position(new List<TopicPartition> { partition }).First();
+                return kafkaHandle.Position(new List<TopicPartition> { partition }).First().Offset;
             }
             catch (TopicPartitionOffsetException e)
             {
                 throw new KafkaException(e.Results[0].Error);
             }
         }
 
-
         /// <inheritdoc/>
         public List<TopicPartitionOffset> OffsetsForTimes(IEnumerable<TopicPartitionTimestamp> timestampsToSearch, TimeSpan timeout)
             // TODO: use a librdkafka queue for this.

src/Confluent.Kafka/IConsumer.cs

@@ -556,24 +556,6 @@ public interface IConsumer<TKey, TValue> : IClient
         ///     Thrown if the request failed.
         /// </exception>
         Offset Position(TopicPartition partition);
-        
-
-        /// <summary>
-        ///     Gets the current position (offset) for the
-        ///     specified topic / partition.
-        ///
-        ///     The offset field of each requested partition
-        ///     will be set to the offset of the last consumed
-        ///     message + 1, or Offset.Unset in case there was
-        ///     no previous message consumed by this consumer.
-        ///     
-        ///     The returned TopicPartitionOffset contains the leader epoch
-        ///     too.
-        /// </summary>
-        /// <exception cref="Confluent.Kafka.KafkaException">
-        ///     Thrown if the request failed.
-        /// </exception>
-        TopicPartitionOffset PositionTopicPartitionOffset(TopicPartition partition);
 
 
         /// <summary>

src/Confluent.Kafka/IConsumerExtensions.cs

@@ -0,0 +1,57 @@
+// Copyright 2018 Confluent Inc.
+//
+// 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.
+//
+// Refer to LICENSE for more information.
+
+using System.Collections.Generic;
+using System.Linq;
+
+
+namespace Confluent.Kafka
+{
+    /// <summary>
+    /// Common extension methods for <see cref="IConsumer{TKey, TValue}"/> implementations.
+    /// </summary>
+    public static class IConsumerExtensions
+    {
+        /// <summary>
+        ///     Gets the current position (offset) for the
+        ///     specified topic / partition.
+        ///
+        ///     The offset field of each requested partition
+        ///     will be set to the offset of the last consumed
+        ///     message + 1, or Offset.Unset in case there was
+        ///     no previous message consumed by this consumer.
+        ///     
+        ///     The returned TopicPartitionOffset contains the leader epoch
+        ///     too.
+        /// </summary>
+        /// <exception cref="Confluent.Kafka.KafkaException">
+        ///     Thrown if the request failed.
+        /// </exception>
+        public static TopicPartitionOffset PositionTopicPartitionOffset<TKey, TValue>(
+            this IConsumer<TKey, TValue> consumer,
+            TopicPartition partition)
+        {
+            try
+            {
+                return consumer.Handle.LibrdkafkaHandle.Position(new List<TopicPartition> { partition }).First();
+            }
+            catch (TopicPartitionOffsetException e)
+            {
+                throw new KafkaException(e.Results[0].Error);
+            }
+        }
+    }
+}