make new offset behavior standard

* remove config setting for NewCommitBehavior
* use committed offset as consumed offset when constructing PartitionTopicInfo as having this being committed - 1 is causing errors on first fetch... apparently this should be the same as the first fetched offset or it will indicate potential data loss.

Author: ducas

src/KafkaNET.Library/Cfg/ConsumerConfiguration.cs

@@ -76,8 +76,6 @@ public class ConsumerConfiguration
 
         public const int DefaultConsumeGroupFindNewLeaderSleepIntervalMs = 2000;
 
-        public const bool DefaultNewCommitBehavior = false;
-
         #region Constructor
         public ConsumerConfiguration()
         {
@@ -101,7 +99,6 @@ public ConsumerConfiguration()
             this.Verbose = false;
             this.ConsumeGroupRebalanceRetryIntervalMs = DefaultConsumeGroupRebalanceRetryIntervalMs;
             this.ConsumeGroupFindNewLeaderSleepIntervalMs = DefaultConsumeGroupFindNewLeaderSleepIntervalMs;
-            this.NewCommitBehavior = DefaultNewCommitBehavior;
         }
 
         private string GetHostName()
@@ -354,13 +351,6 @@ public int MaxFetchSize
         /// </summary>
         public int MaxFetchFactor { get; set; }
 
-        /// <summary>
-        /// Whether to use the new offset commit behaviour of the next fetch offset
-        /// or the traditional behaviour of the consumed offset.
-        /// Default = false.
-        /// </summary>
-        public bool NewCommitBehavior { get; set; }
-
         #endregion
 
         private static void Validate(ConsumerConfigurationSection config)

src/KafkaNET.Library/Consumers/ZookeeperConsumerConnector.cs

@@ -142,7 +142,7 @@ public void CommitOffsets()
                     var topicDirs = new ZKGroupTopicDirs(this.config.GroupId, topic.Key);
                     foreach (KeyValuePair<int, PartitionTopicInfo> partition in topic.Value)
                     {
-                        var newOffset = partition.Value.ConsumeOffset + (config.NewCommitBehavior ? 1 : 0);
+                        var newOffset = partition.Value.ConsumeOffset + 1;
                         try
                         {
                             if (partition.Value.ConsumeOffsetValid)

src/KafkaNET.Library/ZooKeeperIntegration/Listeners/ZKRebalancerListener.cs

@@ -520,20 +520,19 @@ private void AddPartitionTopicInfo(ZKGroupTopicDirs topicDirs, string partition,
             {
                 offsetCommited = long.Parse(offsetString);
                 long latestOffset = this.EarliestOrLatestOffset(topic, leader, partitionId, OffsetRequest.LatestTime);
-                offset = config.NewCommitBehavior ? offsetCommited : offsetCommited + 1;
+                offset = offsetCommited;
                 offset = Math.Min(offset, latestOffset);
                 Logger.InfoFormat("Final offset {0} for topic {1} partition {2} OffsetCommited {3} latestOffset {4}"
                     , offset, topic, partition, offsetCommited, latestOffset);
             }
 
-            var offsetConsumed = config.NewCommitBehavior ? offsetCommited - 1 : offsetCommited;
             var queue = this.queues[new Tuple<string, string>(topic, consumerThreadId)];
             var partTopicInfo = new PartitionTopicInfo(
                 topic,
                 leader,
                 partitionId,
                 queue,
-                offsetConsumed,
+                offset,
                 offset,
                 offset,
                 this.config.FetchSize,