Fix offsets not being committed bug

Author: kmcclellan

Confluent.Kafka.Dataflow/Confluent.Kafka.Dataflow.csproj

@@ -22,6 +22,8 @@
     <PackageTags>kafka;confluent;dataflow</PackageTags>
     <RepositoryType>git</RepositoryType>
     <RepositoryUrl>https://github.com/kmcclellan/kafka-dataflow.git</RepositoryUrl>
+    <IncludeSymbols>true</IncludeSymbols>
+    <SymbolPackageFormat>snupkg</SymbolPackageFormat>
   </PropertyGroup>
 
   <ItemGroup>

Confluent.Kafka.Dataflow/Internal/ProducerBlockFactory.cs

@@ -56,13 +56,16 @@ public ITargetBlock<IEnumerable<TopicPartitionOffset>> GetOffsetTarget(
             return this.GetBatchTarget<TopicPartitionOffset>(
                 offsets =>
                 {
-                    var commitOffsets = new List<TopicPartitionOffset>();
+                    // Track the latest offset per partition.
+                    // (Client seems to unable to handle a large number of offsets).
+                    var commitOffsets = new Dictionary<TopicPartition, TopicPartitionOffset>();
                     foreach (var offset in offsets)
                     {
-                        commitOffsets.Add(new TopicPartitionOffset(offset.TopicPartition, offset.Offset + 1));
+                        commitOffsets[offset.TopicPartition] = 
+                            new TopicPartitionOffset(offset.TopicPartition, offset.Offset + 1);
                     }
 
-                    this.producer.SendOffsetsToTransaction(commitOffsets, consumerGroup, Timeout.InfiniteTimeSpan);
+                    this.producer.SendOffsetsToTransaction(commitOffsets.Values, consumerGroup, Timeout.InfiniteTimeSpan);
                 },
                 options);
         }