From ab5c14e678c3c66709f48874b71078762cb8dbd5 Mon Sep 17 00:00:00 2001
From: Jules Bovet <j.bovet@criteo.com>
Date: Thu, 31 Jan 2019 17:49:32 +0100
Subject: [PATCH 01/15] [Kafka] Introduce partition selector with message key

Currently, there is only one partition selection mode, which is
the round robin one.

There's a need to introduce a new selection strategy depending on
the keys of messages to determine the partition the messages should
go to. It would allow to compact the same messages and reduce heavily
the number of messages.

This new partition selection strategy guarantees that messages with
the same key will be sent to the same partition, given the same
number of available partitions. It is not using a consistent hashing
though, which means that if new nodes are introduced or old ones are
removed from the cluster, new messages with the same key as old
messages will not be sent to the same partition.

JIRA: WBSC-3909

Change-Id: Ie75b40123fd90550aecafd9c33d2b82952470469
---
 .../kafka-sharp.UTest/Kafka.UTest.csproj      |  1 +
 kafka-sharp/kafka-sharp.UTest/TestGeneral.cs  |  8 ++
 .../TestMessageKeyPartitionSelection.cs       | 86 +++++++++++++++++++
 .../kafka-sharp.UTest/TestPartitioner.cs      | 44 ++++++----
 kafka-sharp/kafka-sharp.UTest/TestRouter.cs   |  2 +-
 .../kafka-sharp.UTest/TestSerialization.cs    |  3 -
 kafka-sharp/kafka-sharp/Kafka.csproj          |  8 +-
 kafka-sharp/kafka-sharp/Protocol/Message.cs   |  7 +-
 .../kafka-sharp/Public/ClusterClient.cs       |  3 +-
 .../kafka-sharp/Public/Configuration.cs       |  7 ++
 .../Public/PartitionSelectionConfig.cs        | 59 +++++++++++++
 .../kafka-sharp/Public/Serialization.cs       |  2 +-
 .../PartitionSelection/IPartitionSelection.cs | 23 +++++
 .../MessageKeyPartitionSelection.cs           | 71 +++++++++++++++
 .../PartitionSelection/PartitionSelector.cs   | 52 +++++++++++
 .../RoundRobinPartitionSelection.cs           | 63 ++++++++++++++
 .../kafka-sharp/Routing/PartitionSelector.cs  | 79 -----------------
 .../kafka-sharp/Routing/ProducerRouter.cs     | 13 +--
 18 files changed, 418 insertions(+), 113 deletions(-)
 create mode 100644 kafka-sharp/kafka-sharp.UTest/TestMessageKeyPartitionSelection.cs
 create mode 100644 kafka-sharp/kafka-sharp/Public/PartitionSelectionConfig.cs
 create mode 100644 kafka-sharp/kafka-sharp/Routing/PartitionSelection/IPartitionSelection.cs
 create mode 100644 kafka-sharp/kafka-sharp/Routing/PartitionSelection/MessageKeyPartitionSelection.cs
 create mode 100644 kafka-sharp/kafka-sharp/Routing/PartitionSelection/PartitionSelector.cs
 create mode 100644 kafka-sharp/kafka-sharp/Routing/PartitionSelection/RoundRobinPartitionSelection.cs
 delete mode 100644 kafka-sharp/kafka-sharp/Routing/PartitionSelector.cs

diff --git a/kafka-sharp/kafka-sharp.UTest/Kafka.UTest.csproj b/kafka-sharp/kafka-sharp.UTest/Kafka.UTest.csproj
index 2ec4f32..75adf68 100644
--- a/kafka-sharp/kafka-sharp.UTest/Kafka.UTest.csproj
+++ b/kafka-sharp/kafka-sharp.UTest/Kafka.UTest.csproj
@@ -69,6 +69,7 @@
     <Reference Include="System.Xml" />
   </ItemGroup>
   <ItemGroup>
+    <Compile Include="TestMessageKeyPartitionSelection.cs" />
     <Compile Include="TestVarIntConverter.cs" />
     <Compile Include="TestBatching.cs" />
     <Compile Include="TestClient.cs" />
diff --git a/kafka-sharp/kafka-sharp.UTest/TestGeneral.cs b/kafka-sharp/kafka-sharp.UTest/TestGeneral.cs
index 92f7ebe..d07a709 100644
--- a/kafka-sharp/kafka-sharp.UTest/TestGeneral.cs
+++ b/kafka-sharp/kafka-sharp.UTest/TestGeneral.cs
@@ -12,6 +12,8 @@ namespace tests_kafka_sharp
     [TestFixture]
     internal class TestGeneral
     {
+        private static ISerializer Serializer = new StringSerializer();
+
         private static ClusterClient InitCluster(Configuration configuration, ILogger logger, MetadataResponse metadata, bool forceErrors = false, bool forceConnectionErrors = false, int responseDelay = 0)
         {
             TestData.Reset();
@@ -36,6 +38,7 @@ public void TestOneProduce()
                 ErrorStrategy = ErrorStrategy.Discard,
                 Seeds = "localhost:1,localhost:2,localhost:3"
             };
+            configuration.SerializationConfig.SetDefaultSerializers(Serializer, Serializer);
             const int expectedLatency = 5;
             var cluster = InitCluster(configuration, logger, TestData.TestMetadataResponse, forceErrors: false, forceConnectionErrors: false, responseDelay: expectedLatency);
 
@@ -106,6 +109,7 @@ public void TestMultipleProduce()
                 ErrorStrategy = ErrorStrategy.Discard,
                 Seeds = "localhost:1,localhost:2,localhost:3"
             };
+            configuration.SerializationConfig.SetDefaultSerializers(Serializer, Serializer);
             TestMultipleProduce(configuration);
         }
 
@@ -120,6 +124,7 @@ public void TestMultipleProduceGlobalAccumulator()
                 Seeds = "localhost:1,localhost:2,localhost:3",
                 BatchStrategy = BatchStrategy.Global
             };
+            configuration.SerializationConfig.SetDefaultSerializers(Serializer, Serializer);
             TestMultipleProduce(configuration);
         }
 
@@ -134,6 +139,7 @@ public void TestMultipleProduceConcurrencyOne()
                 Seeds = "localhost:1,localhost:2,localhost:3",
                 TaskScheduler = new ActionBlockTaskScheduler(1)
             };
+            configuration.SerializationConfig.SetDefaultSerializers(Serializer, Serializer);
             TestMultipleProduce(configuration);
         }
 
@@ -196,6 +202,7 @@ public void TestMultipleProduceWithNetworkErrorsAndRetry()
                 MaxSuccessiveNodeErrors = 10,
                 Seeds = "localhost:1,localhost:2,localhost:3"
             };
+            configuration.SerializationConfig.SetDefaultSerializers(Serializer, Serializer);
             var cluster = InitCluster(configuration, logger, TestData.TestMetadataResponse, forceErrors: false, forceConnectionErrors: true);
 
             cluster.Produce("topic1", "key", "value");
@@ -245,6 +252,7 @@ public void TestBigShake()
                 ErrorStrategy = ErrorStrategy.Retry,
                 Seeds = "localhost:1,localhost:2,localhost:3"
             };
+            configuration.SerializationConfig.SetDefaultSerializers(Serializer, Serializer);
             var cluster = InitCluster(configuration, logger, TestData.TestMetadataResponse, forceErrors: true, forceConnectionErrors: true);
 
             cluster.Produce("topic1", "key", "value");
diff --git a/kafka-sharp/kafka-sharp.UTest/TestMessageKeyPartitionSelection.cs b/kafka-sharp/kafka-sharp.UTest/TestMessageKeyPartitionSelection.cs
new file mode 100644
index 0000000..2beaa4a
--- /dev/null
+++ b/kafka-sharp/kafka-sharp.UTest/TestMessageKeyPartitionSelection.cs
@@ -0,0 +1,86 @@
+using System;
+using System.Collections.Generic;
+using Kafka.Cluster;
+using Kafka.Protocol;
+using Kafka.Public;
+using Kafka.Routing;
+using Kafka.Routing.PartitionSelection;
+using NUnit.Framework;
+
+namespace tests_kafka_sharp
+{
+    internal class TestMessageKeyPartitionSelection
+    {
+        private static readonly ISerializer Serializer = new StringSerializer();
+        private static readonly RoundRobinPartitionSelection RoundRobinPartitionSelection = new RoundRobinPartitionSelection();
+
+        [Test]
+        public void Test_MessageKeyPartitionSelection_Is_Consistent()
+        {
+            var nodeMock = new NodeMock();
+            var partitions = new[]
+            {
+                new Partition {Id = 0, Leader = nodeMock},
+                new Partition {Id = 1, Leader = nodeMock},
+                new Partition {Id = 2, Leader = nodeMock},
+            };
+            var partitionStrategy = new MessageKeyPartitionSelection(Serializer, RoundRobinPartitionSelection);
+            var partitioner = new PartitionSelector(partitionStrategy);
+            var message1 = ProduceMessage.New(string.Empty, Partitions.Any, new Message { Key = "ThisIsMyKey" }, new DateTime());
+            var message2 = ProduceMessage.New(string.Empty, Partitions.Any, new Message { Key = "ThisIsMyOtherKey" }, new DateTime());
+
+            var expectedPartition1 = partitioner.GetPartition(message1, partitions);
+            var expectedPartition2 = partitioner.GetPartition(message2, partitions);
+            for (var i = 0; i < 300; i++)
+            {
+                var currentPartition1 = partitioner.GetPartition(message1, partitions);
+                var currentPartition2 = partitioner.GetPartition(message2, partitions);
+                Assert.AreEqual(expectedPartition1.Id, currentPartition1.Id);
+                Assert.AreEqual(expectedPartition2.Id, currentPartition2.Id);
+            }
+        }
+
+        [Test]
+        public void Test_MessageKeyPartitionSelection_Fallbacks_To_RoundRobin_If_MessageKey_Null()
+        {
+            var nodeMock = new NodeMock();
+            var partitions = new[]
+            {
+                new Partition {Id = 0, Leader = nodeMock},
+                new Partition {Id = 1, Leader = nodeMock},
+                new Partition {Id = 2, Leader = nodeMock},
+            };
+            var partitionStrategy = new MessageKeyPartitionSelection(Serializer, RoundRobinPartitionSelection);
+            var partitioner = new PartitionSelector(partitionStrategy);
+            var message = ProduceMessage.New(string.Empty, Partitions.Any, new Message { Key = null }, new DateTime());
+
+            var partition = partitioner.GetPartition(message, partitions);
+            Assert.IsTrue(partition.Id != Partition.None.Id);
+        }
+
+        [TestCase(0)]
+        [TestCase(1)]
+        [TestCase(2)]
+        public void Test_MessageKeyPartitionSelection_Fallbacks_To_RoundRobin_If_Partition_Blacklisted(int partitionIdBlacklisted)
+        {
+            var nodeMock = new NodeMock();
+            var partitions = new[]
+            {
+                new Partition {Id = 0, Leader = nodeMock},
+                new Partition {Id = 1, Leader = nodeMock},
+                new Partition {Id = 2, Leader = nodeMock},
+            };
+            var blacklistedPartitions = new Dictionary<int, DateTime> { { partitionIdBlacklisted, DateTime.MaxValue } };
+            var partitionStrategy = new MessageKeyPartitionSelection(Serializer, RoundRobinPartitionSelection);
+            var partitioner = new PartitionSelector(partitionStrategy);
+            var message = ProduceMessage.New(string.Empty, Partitions.Any, new Message { Key = "ThisIsMyKey" }, new DateTime());
+
+            for (var i = 0; i < 300; i++)
+            {
+                var partition = partitioner.GetPartition(message, partitions, blacklistedPartitions);
+                Assert.IsTrue(partition.Id != Partition.None.Id);
+                Assert.IsTrue(partition.Id != partitionIdBlacklisted);
+            }
+        }
+    }
+}
diff --git a/kafka-sharp/kafka-sharp.UTest/TestPartitioner.cs b/kafka-sharp/kafka-sharp.UTest/TestPartitioner.cs
index 9b2093c..3f1a78d 100644
--- a/kafka-sharp/kafka-sharp.UTest/TestPartitioner.cs
+++ b/kafka-sharp/kafka-sharp.UTest/TestPartitioner.cs
@@ -1,8 +1,11 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using Kafka.Cluster;
+using Kafka.Protocol;
 using Kafka.Public;
 using Kafka.Routing;
+using Kafka.Routing.PartitionSelection;
 using NUnit.Framework;
 
 namespace tests_kafka_sharp
@@ -27,13 +30,16 @@ public void TestRoundRobinPartitionAssign(int delay)
                     new Partition {Id = 3, Leader = nodeMock},
                     new Partition {Id = 4, Leader = nodeMock},
                 };
-            var partitioner = new PartitionSelector(delay);
+            var partitionStrategy = new RoundRobinPartitionSelection(delay);
+            var partitioner = new PartitionSelector(partitionStrategy);
             delay = delay <= 0 ? 1 : delay;
             foreach (var partition in partitions)
             {
                 for (var j = 0; j < delay; ++j)
                 {
-                    Assert.AreEqual(partition.Id, partitioner.GetPartition(Partitions.Any, partitions).Id);
+                    Assert.AreEqual(partition.Id, partitioner
+                        .GetPartition(ProduceMessage.New(string.Empty, Partitions.Any, new Message(), new DateTime()), partitions)
+                        .Id);
                 }
             }
         }
@@ -42,8 +48,10 @@ public void TestRoundRobinPartitionAssign(int delay)
         public void TestRoundRobinPartitionAssignNoPartitionReturnsNone()
         {
             var partitions = new Partition[0];
-            var partitioner = new PartitionSelector();
-            Assert.AreEqual(0, Partition.None.CompareTo(partitioner.GetPartition(Partitions.Any, partitions)));
+            var partitionStrategy = new RoundRobinPartitionSelection();
+            var partitioner = new PartitionSelector(partitionStrategy);
+            Assert.AreEqual(0, Partition.None.CompareTo(partitioner.GetPartition(
+                ProduceMessage.New(string.Empty, Partitions.Any, new Message(), new DateTime()), partitions)));
         }
 
         [Test]
@@ -62,15 +70,16 @@ public void TestFilter()
             filter[0] = DateTime.UtcNow;
             filter[2] = DateTime.UtcNow;
             filter[4] = DateTime.UtcNow;
-            var partitioner = new PartitionSelector();
+            var partitionStrategy = new RoundRobinPartitionSelection();
+            var partitioner = new PartitionSelector(partitionStrategy);
 
-            var partition = partitioner.GetPartition(Partitions.Any, partitions, filter);
+            var partition = partitioner.GetPartition(ProduceMessage.New(string.Empty, Partitions.Any, new Message(), new DateTime()),partitions, filter);
             Assert.AreEqual(1, partition.Id);
 
-            partition = partitioner.GetPartition(Partitions.Any, partitions, filter);
+            partition = partitioner.GetPartition(ProduceMessage.New(string.Empty, Partitions.Any, new Message(), new DateTime()), partitions, filter);
             Assert.AreEqual(3, partition.Id);
 
-            partition = partitioner.GetPartition(Partitions.Any, partitions, filter);
+            partition = partitioner.GetPartition(ProduceMessage.New(string.Empty, Partitions.Any, new Message(), new DateTime()), partitions, filter);
             Assert.AreEqual(1, partition.Id);
         }
 
@@ -93,9 +102,10 @@ public void TestRobinPartitionAssignWhenFiltered()
 
             int delay = partitions.Length + 2;
 
-            var partitioner = new PartitionSelector(delay);
+            var partitionStrategy = new RoundRobinPartitionSelection(delay);
+            var partitioner = new PartitionSelector(partitionStrategy);
 
-            var partition = partitioner.GetPartition(Partitions.Any, partitions, filter);
+            var partition = partitioner.GetPartition(ProduceMessage.New(string.Empty, Partitions.Any, new Message(), new DateTime()), partitions, filter);
 
             Assert.AreEqual(1, partition.Id);
 
@@ -128,7 +138,8 @@ public void TestFilterWithHighDelay()
             // Pick a delay greater than the number of partitions
             int delay = partitions.Length + 2;
 
-            var partitioner = new PartitionSelector(delay);
+            var partitionStrategy = new RoundRobinPartitionSelection(delay);
+            var partitioner = new PartitionSelector(partitionStrategy);
 
             var firstBatch = GetPartitions(delay, partitioner, partitions, filter);
 
@@ -152,7 +163,8 @@ public void TestOverflow()
 
             var partitions = Enumerable.Range(0, 10).Select(i => new Partition { Id = i, Leader = nodeMock }).ToArray();
 
-            var partitioner = new PartitionSelector(1, int.MaxValue);
+            var partitionStrategy = new RoundRobinPartitionSelection(delay: 1, startSeed: int.MaxValue);
+            var partitioner = new PartitionSelector(partitionStrategy);
 
             var batch = GetPartitions(partitions.Length, partitioner, partitions, null);
 
@@ -180,12 +192,14 @@ public void TestRoundRobinPartitionWithStartSeed(int startSeed, int delay)
                     new Partition {Id = 3, Leader = nodeMock},
                     new Partition {Id = 4, Leader = nodeMock},
                 };
-            var partitioner = new PartitionSelector(delay, startSeed);
+            var partitionStrategy = new RoundRobinPartitionSelection(delay: delay, startSeed: startSeed);
+            var partitioner = new PartitionSelector(partitionStrategy);
             foreach (var partition in partitions)
             {
                 for (var j = 0; j < delay; ++j)
                 {
-                    Assert.AreEqual((partition.Id + startSeed) % partitions.Length, partitioner.GetPartition(Partitions.Any, partitions).Id);
+                    Assert.AreEqual((partition.Id + startSeed) % partitions.Length, partitioner.GetPartition(
+                        ProduceMessage.New(string.Empty, Partitions.Any, new Message(), new DateTime()), partitions).Id);
                 }
             }
         }
@@ -196,7 +210,7 @@ private static List<Partition> GetPartitions(int count, PartitionSelector partit
 
             for (int i = 0; i < count; i++)
             {
-                result.Add(partitioner.GetPartition(Partitions.Any, partitions, filter));
+                result.Add(partitioner.GetPartition(ProduceMessage.New(string.Empty, Partitions.Any, new Message(), new DateTime()), partitions, filter));
             }
 
             return result;
diff --git a/kafka-sharp/kafka-sharp.UTest/TestRouter.cs b/kafka-sharp/kafka-sharp.UTest/TestRouter.cs
index 9b836d2..b8a2b45 100644
--- a/kafka-sharp/kafka-sharp.UTest/TestRouter.cs
+++ b/kafka-sharp/kafka-sharp.UTest/TestRouter.cs
@@ -183,7 +183,7 @@ public void TestSerializeOnProduce()
                     n.Produce(
                         It.Is<ProduceMessage>(
                             p =>
-                                p.Message.SerializedKeyValue != null && p.Message.Key == null && p.Message.Value == null)));
+                                p.Message.SerializedKeyValue != null && p.Message.Key != null && p.Message.Value == null)));
         }
 
         [Test]
diff --git a/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs b/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs
index 0434d6a..2fdd9a7 100644
--- a/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs
+++ b/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs
@@ -81,7 +81,6 @@ public void TestSerializeOneMessageWithPreserializedKeyValue()
         {
             var message = new Message { Key = Key, Value = Value };
             message.SerializeKeyValue(new ReusableMemoryStream(null), new Tuple<ISerializer, ISerializer>(null, null));
-            Assert.IsNull(message.Key);
             Assert.IsNull(message.Value);
             Assert.IsNotNull(message.SerializedKeyValue);
             TestSerializeOneMessageCommon(message);
@@ -128,7 +127,6 @@ public void TestSerializeOneMessageIMemorySerializableWithPreserializedKeyValue(
         {
             var message = new Message { Key = new SimpleSerializable(Key), Value = new SimpleSerializable(Value) };
             message.SerializeKeyValue(new ReusableMemoryStream(null), new Tuple<ISerializer, ISerializer>(null, null));
-            Assert.IsNull(message.Key);
             Assert.IsNull(message.Value);
             Assert.IsNotNull(message.SerializedKeyValue);
             TestSerializeOneMessageCommon(message);
@@ -226,7 +224,6 @@ public void TestSerializeOneEmptyMessageWithPreserializedKeyValue(MessageVersion
         {
             var message = new Message { Key = new byte[0], Value = new byte[0] };
             message.SerializeKeyValue(new ReusableMemoryStream(null), new Tuple<ISerializer, ISerializer>(null, null));
-            Assert.IsNull(message.Key);
             Assert.IsNull(message.Value);
             Assert.IsNotNull(message.SerializedKeyValue);
             TestSerializeOneEmptyMessageCommon(message, messageVersion);
diff --git a/kafka-sharp/kafka-sharp/Kafka.csproj b/kafka-sharp/kafka-sharp/Kafka.csproj
index f0b11cb..a3b0f13 100644
--- a/kafka-sharp/kafka-sharp/Kafka.csproj
+++ b/kafka-sharp/kafka-sharp/Kafka.csproj
@@ -120,11 +120,15 @@
     <Compile Include="Public\KafkaConsumer.cs" />
     <Compile Include="Public\KafkaProducer.cs" />
     <Compile Include="Public\KafkaRecord.cs" />
+    <Compile Include="Public\PartitionSelectionConfig.cs" />
     <Compile Include="Public\Serialization.cs" />
     <Compile Include="Public\Statistics.cs" />
     <Compile Include="Routing\ConsumerGroup.cs" />
     <Compile Include="Routing\ConsumerRouter.cs" />
-    <Compile Include="Routing\PartitionSelector.cs" />
+    <Compile Include="Routing\PartitionSelection\IPartitionSelection.cs" />
+    <Compile Include="Routing\PartitionSelection\MessageKeyPartitionSelection.cs" />
+    <Compile Include="Routing\PartitionSelection\PartitionSelector.cs" />
+    <Compile Include="Routing\PartitionSelection\RoundRobinPartitionSelection.cs" />
     <Compile Include="Routing\ProducerRouter.cs" />
     <Compile Include="Routing\RoutingTable.cs" />
   </ItemGroup>
@@ -132,7 +136,7 @@
     <None Include="packages.config" />
   </ItemGroup>
   <Import Project="$(MSBuildToolsPath)\Microsoft.CSharp.targets" />
-  <!-- To modify your build process, add your task inside one of the targets below and uncomment it. 
+  <!-- To modify your build process, add your task inside one of the targets below and uncomment it.
        Other similar extension points exist, see Microsoft.Common.targets.
   <Target Name="BeforeBuild">
   </Target>
diff --git a/kafka-sharp/kafka-sharp/Protocol/Message.cs b/kafka-sharp/kafka-sharp/Protocol/Message.cs
index 3dcbaed..0db2f9d 100644
--- a/kafka-sharp/kafka-sharp/Protocol/Message.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/Message.cs
@@ -31,7 +31,6 @@ public void SerializeKeyValue(ReusableMemoryStream target, Tuple<ISerializer, IS
         {
             SerializedKeyValue = target;
             DoSerializeKeyValue(SerializedKeyValue, serializers);
-            Key = null;
             Value = null;
         }
 
@@ -120,8 +119,7 @@ private void DoSerializeKeyValue(ReusableMemoryStream stream, Tuple<ISerializer,
         private static void SerializeObject(ReusableMemoryStream stream, ISerializer serializer, object theValue)
         {
             // byte[] are just copied
-            var bytes = theValue as byte[];
-            if (bytes != null)
+            if (theValue is byte[] bytes)
             {
                 byte[] array = bytes;
                 BigEndianConverter.Write(stream, array.Length);
@@ -135,8 +133,7 @@ private static void SerializeObject(ReusableMemoryStream stream, ISerializer ser
 
         private static void SerializerWrite(ReusableMemoryStream stream, object m, ISerializer ser)
         {
-            var serializable = m as IMemorySerializable;
-            if (serializable != null)
+            if (m is IMemorySerializable serializable)
             {
                 serializable.Serialize(stream);
             }
diff --git a/kafka-sharp/kafka-sharp/Public/ClusterClient.cs b/kafka-sharp/kafka-sharp/Public/ClusterClient.cs
index 27b9b4b..81b7a86 100644
--- a/kafka-sharp/kafka-sharp/Public/ClusterClient.cs
+++ b/kafka-sharp/kafka-sharp/Public/ClusterClient.cs
@@ -609,8 +609,7 @@ public bool Produce(string topic, object key, object data, ICollection<KafkaReco
                     Key = key,
                     Value = data,
                     Headers = _configuration.Compatibility >= Compatibility.V0_11_0 ? headers : null,
-                    TimeStamp =
-                        _configuration.Compatibility >= Compatibility.V0_10_1 ? Timestamp.ToUnixTimestamp(timestamp) : 0
+                    TimeStamp = _configuration.Compatibility >= Compatibility.V0_10_1 ? Timestamp.ToUnixTimestamp(timestamp) : 0
                 },
                 partition, DateTime.UtcNow.Add(_configuration.MessageTtl));
             return true;
diff --git a/kafka-sharp/kafka-sharp/Public/Configuration.cs b/kafka-sharp/kafka-sharp/Public/Configuration.cs
index ab321fa..f936449 100644
--- a/kafka-sharp/kafka-sharp/Public/Configuration.cs
+++ b/kafka-sharp/kafka-sharp/Public/Configuration.cs
@@ -349,6 +349,13 @@ public class Configuration
         /// Switch to earliest offset or latest offset.
         /// </summary>
         public Offset OffsetOutOfRangeStrategy = Offset.Earliest;
+
+        /// <summary>
+        /// Class holding information about the partition selection strategy for
+        /// each topic. It uses the default Round Robin strategy if none is set
+        /// for a given topic.
+        /// </summary>
+        public PartitionSelectionConfig PartitionSelectionConfig = new PartitionSelectionConfig();
     }
 
     public enum Offset
diff --git a/kafka-sharp/kafka-sharp/Public/PartitionSelectionConfig.cs b/kafka-sharp/kafka-sharp/Public/PartitionSelectionConfig.cs
new file mode 100644
index 0000000..eff5985
--- /dev/null
+++ b/kafka-sharp/kafka-sharp/Public/PartitionSelectionConfig.cs
@@ -0,0 +1,59 @@
+using System.Collections.Generic;
+using Kafka.Routing.PartitionSelection;
+
+namespace Kafka.Public
+{
+    public enum PartitionSelectionStrategy
+    {
+        /// <summary>
+        /// Round Robin between all available partitions.
+        /// </summary>
+        RoundRobin,
+
+        /// <summary>
+        /// Choose the partition depending on the message's key. It guarantees that given the same
+        /// number of available partitions, messages with the same key will go to the same partition.
+        /// If the message's key is null, it fall backs to the round robin strategy.
+        /// If the chosen partition is blacklisted, messages to that partition will actually be sent
+        /// to another partition using the round robin strategy.
+        /// </summary>
+        MessageKey,
+    }
+
+    public sealed class PartitionSelectionConfig
+    {
+        private readonly Dictionary<string, PartitionSelectionStrategy> _selectionStrategyByTopic;
+        private PartitionSelectionStrategy _defaultStrategy = PartitionSelectionStrategy.RoundRobin;
+
+        public PartitionSelectionConfig()
+        {
+            _selectionStrategyByTopic = new Dictionary<string, PartitionSelectionStrategy>();
+        }
+
+        public void SetDefaultPartitionSelectionStrategy(PartitionSelectionStrategy strategy)
+        {
+            _defaultStrategy = strategy;
+        }
+
+        public void SetPartitionSelectionStrategyForTopic(string topic, PartitionSelectionStrategy strategy)
+        {
+            _selectionStrategyByTopic.Add(topic, strategy);
+        }
+
+        internal IPartitionSelection GetPartitionSelectionForTopic(string topic, int delay, int startSeed, ISerializer keySerializer)
+        {
+            if (!_selectionStrategyByTopic.TryGetValue(topic, out var selectionStrategy))
+            {
+                selectionStrategy = _defaultStrategy;
+            }
+
+            var roundRobinSelection = new RoundRobinPartitionSelection(delay, startSeed);
+            if (selectionStrategy == PartitionSelectionStrategy.RoundRobin)
+            {
+                return roundRobinSelection;
+            }
+
+            return new MessageKeyPartitionSelection(keySerializer, roundRobinSelection);
+        }
+    }
+}
diff --git a/kafka-sharp/kafka-sharp/Public/Serialization.cs b/kafka-sharp/kafka-sharp/Public/Serialization.cs
index dc4a028..61538d8 100644
--- a/kafka-sharp/kafka-sharp/Public/Serialization.cs
+++ b/kafka-sharp/kafka-sharp/Public/Serialization.cs
@@ -236,7 +236,7 @@ public long SerializedSize(object input)
     }
 
     /// <summary>
-    /// A desrializer for strings.
+    /// A deserializer for strings.
     /// </summary>
     public sealed class StringDeserializer : IDeserializer
     {
diff --git a/kafka-sharp/kafka-sharp/Routing/PartitionSelection/IPartitionSelection.cs b/kafka-sharp/kafka-sharp/Routing/PartitionSelection/IPartitionSelection.cs
new file mode 100644
index 0000000..88d2232
--- /dev/null
+++ b/kafka-sharp/kafka-sharp/Routing/PartitionSelection/IPartitionSelection.cs
@@ -0,0 +1,23 @@
+using System;
+using System.Collections.Generic;
+using Kafka.Cluster;
+
+namespace Kafka.Routing.PartitionSelection
+{
+    /// <summary>
+    /// A class responsible of selecting a partition to send a kafka message.
+    /// </summary>
+    internal interface IPartitionSelection
+    {
+        /// <summary>
+        /// Select a partition to send the message to, taking into account that some partitions might
+        /// be temporary blacklisted.
+        /// </summary>
+        /// <param name="produceMessage">Encapsulation of a message sent to Kafka brokers. It contains additional
+        ///   information like the RequiredPartition that indicates we want to target a specific partition</param>
+        /// <param name="partitions">The list of all available partitions</param>
+        /// <param name="blacklist">Dictionary of partition ids that are temporary blacklisted</param>
+        /// <returns>The partition the message will be sent to</returns>
+        Partition GetPartition(ProduceMessage produceMessage, Partition[] partitions, IReadOnlyDictionary<int, DateTime> blacklist);
+    }
+}
diff --git a/kafka-sharp/kafka-sharp/Routing/PartitionSelection/MessageKeyPartitionSelection.cs b/kafka-sharp/kafka-sharp/Routing/PartitionSelection/MessageKeyPartitionSelection.cs
new file mode 100644
index 0000000..01a5cce
--- /dev/null
+++ b/kafka-sharp/kafka-sharp/Routing/PartitionSelection/MessageKeyPartitionSelection.cs
@@ -0,0 +1,71 @@
+using System;
+using System.Collections.Generic;
+using Kafka.Cluster;
+using Kafka.Common;
+using Kafka.Public;
+using Kafka.Public.Loggers;
+
+namespace Kafka.Routing.PartitionSelection
+{
+    internal class MessageKeyPartitionSelection : IPartitionSelection
+    {
+        private static readonly ILogger Logger = LoggerManager.Instance;
+        private const int LogSamplingPercentage = 99;
+
+        private readonly ISerializer _keySerializer;
+        private readonly IPartitionSelection _roundRobinSelection;
+        private readonly Pool<ReusableMemoryStream> _messageKeyBuffersPool;
+        private readonly Random _rnd;
+
+        public MessageKeyPartitionSelection(ISerializer keySerializer, IPartitionSelection roundRobinSelection)
+        {
+            _keySerializer = keySerializer;
+            _roundRobinSelection = roundRobinSelection;
+            _rnd = new Random();
+            _messageKeyBuffersPool = new Pool<ReusableMemoryStream>(
+                limit: 100,
+                constructor: () => new ReusableMemoryStream(_messageKeyBuffersPool),
+                clearAction: (stream, reused) =>
+                {
+                    if (reused)
+                        stream.SetLength(0);
+                });
+        }
+
+        public Partition GetPartition(ProduceMessage produceMessage, Partition[] partitions, IReadOnlyDictionary<int, DateTime> blacklist)
+        {
+            var partitionId = GetPartitionIdFromKey(produceMessage.Message.Key, partitions.Length);
+            if (partitionId != Partition.None.Id && !blacklist.ContainsKey(partitionId))
+            {
+                return partitions[partitionId];
+            }
+
+            return _roundRobinSelection.GetPartition(produceMessage, partitions, blacklist);
+        }
+
+        /// <summary>
+        /// Compute the partition id given the message key and the number of partitions
+        /// Do a simple Crc32 of the key, modulo the number of partitions. This is the exact same
+        /// algorithm in librdkafka
+        /// </summary>
+        private int GetPartitionIdFromKey(object messageKey, int partitionsLength)
+        {
+            if (messageKey != null)
+            {
+                var memoryStream = _messageKeyBuffersPool.Reserve();
+                _keySerializer.Serialize(messageKey, memoryStream);
+                var partitionId = Crc32.Compute(memoryStream, 0, memoryStream.Length) % partitionsLength;
+                _messageKeyBuffersPool.Release(memoryStream);
+                return (int) partitionId;
+            }
+
+            if (_rnd.Next(0, 100) >= LogSamplingPercentage)
+            {
+                Logger.LogError($"{typeof(MessageKeyPartitionSelection)} cannot determine partition as message's "
+                    + "key is null. Falling back to round robin selection");
+            }
+
+            return Partition.None.Id;
+        }
+    }
+}
diff --git a/kafka-sharp/kafka-sharp/Routing/PartitionSelection/PartitionSelector.cs b/kafka-sharp/kafka-sharp/Routing/PartitionSelection/PartitionSelector.cs
new file mode 100644
index 0000000..c59b54e
--- /dev/null
+++ b/kafka-sharp/kafka-sharp/Routing/PartitionSelection/PartitionSelector.cs
@@ -0,0 +1,52 @@
+// 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
+
+using System;
+using System.Collections.Generic;
+using Kafka.Cluster;
+using Kafka.Public;
+
+namespace Kafka.Routing.PartitionSelection
+{
+    /// <summary>
+    /// Select a partition from an array of Partition.
+    /// </summary>
+    internal class PartitionSelector
+    {
+        private readonly IPartitionSelection _partitionSelection;
+        private static readonly IReadOnlyDictionary<int, DateTime> EmptyBlackList = new Dictionary<int, DateTime>();
+
+        public PartitionSelector(IPartitionSelection partitionSelection)
+        {
+            _partitionSelection = partitionSelection;
+        }
+
+        /// <summary>
+        /// Get the partition the message will be sent to.
+        /// Actual selection strategy depends on the IPartitionSelection implementation
+        /// that was chosen for this topic.
+        /// </summary>
+        /// <param name="produceMessage">The ProduceMessage to send</param>
+        /// <param name="partitions">List of all available partitions</param>
+        /// <param name="blacklist">Dictionary of partition ids that are currently blacklisted</param>
+        /// <returns></returns>
+        public Partition GetPartition(ProduceMessage produceMessage, Partition[] partitions, IReadOnlyDictionary<int, DateTime> blacklist = null)
+        {
+            blacklist = blacklist ?? EmptyBlackList;
+
+            switch (produceMessage.RequiredPartition)
+            {
+                case Partitions.None:
+                    return Partition.None;
+
+                case Partitions.Any:
+                case Partitions.All:
+                    return _partitionSelection.GetPartition(produceMessage, partitions, blacklist);
+
+                default:
+                    var found = Array.BinarySearch(partitions, new Partition { Id = produceMessage.RequiredPartition });
+                    return found >= 0 ? partitions[found] : Partition.None;
+            }
+        }
+    }
+}
diff --git a/kafka-sharp/kafka-sharp/Routing/PartitionSelection/RoundRobinPartitionSelection.cs b/kafka-sharp/kafka-sharp/Routing/PartitionSelection/RoundRobinPartitionSelection.cs
new file mode 100644
index 0000000..b72b985
--- /dev/null
+++ b/kafka-sharp/kafka-sharp/Routing/PartitionSelection/RoundRobinPartitionSelection.cs
@@ -0,0 +1,63 @@
+using System;
+using System.Collections.Generic;
+using Kafka.Cluster;
+
+namespace Kafka.Routing.PartitionSelection
+{
+    internal class RoundRobinPartitionSelection : IPartitionSelection
+    {
+        private long _next;
+        private int _cursor;
+        private readonly int _delay;
+
+        public RoundRobinPartitionSelection(int delay = 1, int startSeed = 0)
+        {
+            _delay = delay <= 0 ? 1 : delay;
+            _next = startSeed <= 0 ? 0L : startSeed;
+        }
+
+        public Partition GetPartition(ProduceMessage produceMessage, Partition[] partitions, IReadOnlyDictionary<int, DateTime> blacklist)
+        {
+            return GetRandomPartition(partitions, blacklist);
+        }
+
+        private Partition GetRandomPartition(Partition[] partitions, IReadOnlyDictionary<int, DateTime> blacklist)
+        {
+            for (var retryCount = 0; retryCount < partitions.Length; retryCount++)
+            {
+                var partition = GetActivePartition(partitions);
+
+                if (blacklist.ContainsKey(partition.Id))
+                {
+                    SelectNextPartition();
+                    continue;
+                }
+
+                _cursor++;
+
+                if (_cursor >= _delay)
+                {
+                    // Round-robin threshold met, switch to next partition for the next call
+                    SelectNextPartition();
+                }
+
+                return partition;
+            }
+
+            return Partition.None;
+        }
+
+        private Partition GetActivePartition(Partition[] partitions)
+        {
+            return partitions[_next % partitions.Length];
+        }
+
+        private void SelectNextPartition()
+        {
+            _next++;
+
+            // Reset the hit count for current partition
+            _cursor = 0;
+        }
+    }
+}
diff --git a/kafka-sharp/kafka-sharp/Routing/PartitionSelector.cs b/kafka-sharp/kafka-sharp/Routing/PartitionSelector.cs
deleted file mode 100644
index 6aeffd4..0000000
--- a/kafka-sharp/kafka-sharp/Routing/PartitionSelector.cs
+++ /dev/null
@@ -1,79 +0,0 @@
-// 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
-
-using System;
-using System.Collections.Generic;
-using Kafka.Public;
-
-namespace Kafka.Routing
-{
-    /// <summary>
-    /// Select a partition from an array of Partition.
-    /// </summary>
-    internal class PartitionSelector
-    {
-        private long _next;
-        private int _cursor;
-        private readonly int _delay;
-        private static readonly IReadOnlyDictionary<int, DateTime> EmptyBlackList = new Dictionary<int, DateTime>();
-
-        public PartitionSelector(int delay = 1, int startSeed = 0)
-        {
-            _delay = delay <= 0 ? 1 : delay;
-            _next = startSeed <= 0 ? 0L : startSeed;
-        }
-
-        public Partition GetPartition(int partitionId, Partition[] partitions, IReadOnlyDictionary<int, DateTime> blacklist = null)
-        {
-            blacklist = blacklist ?? EmptyBlackList;
-
-            switch (partitionId)
-            {
-                case Partitions.None:
-                    return Partition.None;
-
-                case Partitions.Any:
-                case Partitions.All:
-                    for (int retryCount = 0; retryCount < partitions.Length; retryCount++)
-                    {
-                        var partition = GetActivePartition(partitions);
-
-                        if (blacklist.ContainsKey(partition.Id))
-                        {
-                            SelectNextPartition();
-                            continue;
-                        }
-
-                        _cursor++;
-
-                        if (_cursor >= _delay)
-                        {
-                            // Round-robin threshold met, switch to next partition for the next call
-                            SelectNextPartition();
-                        }
-
-                        return partition;
-                    }
-
-                    return Partition.None;
-
-                default:
-                    var found = Array.BinarySearch(partitions, new Partition { Id = partitionId });
-                    return found >= 0 ? partitions[found] : Partition.None;
-            }
-        }
-
-        private Partition GetActivePartition(Partition[] partitions)
-        {
-            return partitions[_next % partitions.Length];
-        }
-
-        private void SelectNextPartition()
-        {
-            _next++;
-
-            // Reset the hit count for current partition
-            _cursor = 0;
-        }
-    }
-}
diff --git a/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs b/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs
index b5650b8..069ea9d 100644
--- a/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs
+++ b/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs
@@ -13,6 +13,7 @@
 using Kafka.Common;
 using Kafka.Protocol;
 using Kafka.Public;
+using Kafka.Routing.PartitionSelection;
 using ICluster = Kafka.Cluster.ICluster;
 
 namespace Kafka.Routing
@@ -563,10 +564,12 @@ private async Task HandleProduceMessage(ProduceMessage produceMessage)
                 return;
             }
 
-            PartitionSelector selector;
-            if (!_partitioners.TryGetValue(topic, out selector))
+            if (!_partitioners.TryGetValue(topic, out var selector))
             {
-                selector = new PartitionSelector(_configuration.NumberOfMessagesBeforeRoundRobin, _randomGenerator.Next());
+                var keySerializer = _configuration.SerializationConfig.GetSerializersForTopic(topic).Item1;
+                var partitionSelectionImpl = _configuration.PartitionSelectionConfig.GetPartitionSelectionForTopic(
+                    topic, _configuration.NumberOfMessagesBeforeRoundRobin, _randomGenerator.Next(), keySerializer);
+                selector = new PartitionSelector(partitionSelectionImpl);
                 _partitioners[topic] = selector;
             }
 
@@ -577,13 +580,13 @@ private async Task HandleProduceMessage(ProduceMessage produceMessage)
                 partitions = _routingTable.GetPartitions(topic);
             }
 
-            var partition = selector.GetPartition(produceMessage.RequiredPartition, partitions, GetFilter(topic));
+            var partition = selector.GetPartition(produceMessage, partitions, GetFilter(topic));
 
             if (partition.Id == Partitions.None)
             {
                 // Retry without filters because filtered partitions should be valid, we just wanted to avoid
                 // spamming them while they're being rebalanced.
-                partition = selector.GetPartition(produceMessage.RequiredPartition, partitions);
+                partition = selector.GetPartition(produceMessage, partitions);
 
                 if (partition.Id == Partitions.None)
                 {

From b9925664f4c8f4e4f399748e277cc1adcb4f1079 Mon Sep 17 00:00:00 2001
From: Stanislas POCTHIER <s.pocthier@criteo.com>
Date: Fri, 8 Mar 2019 15:41:17 +0100
Subject: [PATCH 02/15] Open more RawKafkaMessage so that clients can test with
 it

For client of the driver (i.e. the Kafka service) it is useful
to be able to create messages for testing purpose, and to be
able to manually set their partition/topic etc.

Change-Id: Ife4b6e29e0ea4e732f58e9500f085ca06a390312
---
 kafka-sharp/kafka-sharp/Public/KafkaRecord.cs | 15 +++++++++++++++
 1 file changed, 15 insertions(+)

diff --git a/kafka-sharp/kafka-sharp/Public/KafkaRecord.cs b/kafka-sharp/kafka-sharp/Public/KafkaRecord.cs
index 3e8a648..abd8abd 100644
--- a/kafka-sharp/kafka-sharp/Public/KafkaRecord.cs
+++ b/kafka-sharp/kafka-sharp/Public/KafkaRecord.cs
@@ -85,6 +85,21 @@ public int Partition
     /// </summary>
     public class RawKafkaRecord
     {
+        public RawKafkaRecord() { }
+
+        // For testing purpose
+        public RawKafkaRecord(string topic = null, object key = null, object value = null, long offset = 0,
+            long lag = 0, int partition = 0, DateTime? timestamp = null)
+        {
+            Topic = topic;
+            Key = key;
+            Value = value;
+            Offset = offset;
+            Lag = lag;
+            Partition = partition;
+            Timestamp = timestamp.GetValueOrDefault(DateTime.MinValue);
+        }
+
         /// <summary>
         /// The topic of the record.
         /// </summary>

From 27fb5d36231a90ea17408a28d75514a606ca5677 Mon Sep 17 00:00:00 2001
From: Stanislas POCTHIER <s.pocthier@criteo.com>
Date: Thu, 18 Apr 2019 13:52:25 +0200
Subject: [PATCH 03/15] Change format of Info log message

We currently have an issue with kibana where for some messages
the labels are not correctly parsed, and they are considered part
of the message. This messes up with getting logs easily.

I am trying to change this message to see if it fixes the issue.

Change-Id: I45a24df49a1d564a4c8a8dd0453d1fd72a5cc237
---
 kafka-sharp/kafka-sharp/Protocol/Metadata.cs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/kafka-sharp/kafka-sharp/Protocol/Metadata.cs b/kafka-sharp/kafka-sharp/Protocol/Metadata.cs
index a43e312..2e126ca 100644
--- a/kafka-sharp/kafka-sharp/Protocol/Metadata.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/Metadata.cs
@@ -13,7 +13,7 @@ class BrokerMeta : IMemoryStreamSerializable
 
         public override string ToString()
         {
-            return string.Format("[Id:{0} Host:{1} Port:{2}]", Id, Host, Port);
+            return string.Format("(Id:{0} Host:{1} Port:{2})", Id, Host, Port);
         }
 
         public void Deserialize(ReusableMemoryStream stream, object _, Basics.ApiVersion __)

From dcf3c44f188665647954d6d19d7e44d38da7eac9 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Gr=C3=A9goire=20Verdier?= <g.verdier@criteo.com>
Date: Fri, 10 May 2019 10:16:53 +0200
Subject: [PATCH 04/15] Move GetApiVersion to Basics

Change-Id: I7a054467983c93ee307c8b6410e15c4fd03bd964
---
 kafka-sharp/kafka-sharp/Cluster/Node.cs    | 83 ++--------------------
 kafka-sharp/kafka-sharp/Protocol/Basics.cs | 74 +++++++++++++++++++
 2 files changed, 79 insertions(+), 78 deletions(-)

diff --git a/kafka-sharp/kafka-sharp/Cluster/Node.cs b/kafka-sharp/kafka-sharp/Cluster/Node.cs
index b791e62..b660c68 100644
--- a/kafka-sharp/kafka-sharp/Cluster/Node.cs
+++ b/kafka-sharp/kafka-sharp/Cluster/Node.cs
@@ -316,7 +316,7 @@ public Serialization(SerializationConfig serializationConfig, Compatibility comp
 
             public ReusableMemoryStream SerializeMetadataAllRequest(int correlationId)
             {
-                return new TopicRequest().Serialize(_requestPool.Reserve(), correlationId, _clientId, null, GetApiVersion(RequestType.Metadata, _compatibility));
+                return new TopicRequest().Serialize(_requestPool.Reserve(), correlationId, _clientId, null, Basics.GetApiVersion(RequestType.Metadata, _compatibility));
             }
 
             public ReusableMemoryStream SerializeProduceBatch(int correlationId, IEnumerable<IGrouping<string, IGrouping<int, ProduceMessage>>> batch)
@@ -336,7 +336,7 @@ public ReusableMemoryStream SerializeProduceBatch(int correlationId, IEnumerable
                         })
                     })
                 };
-                return produceRequest.Serialize(_requestPool.Reserve(), correlationId, _clientId, _serializationConfig, GetApiVersion(RequestType.BatchedProduce, _compatibility));
+                return produceRequest.Serialize(_requestPool.Reserve(), correlationId, _clientId, _serializationConfig, Basics.GetApiVersion(RequestType.BatchedProduce, _compatibility));
             }
 
             public ReusableMemoryStream SerializeFetchBatch(int correlationId, IEnumerable<IGrouping<string, FetchMessage>> batch)
@@ -357,7 +357,7 @@ public ReusableMemoryStream SerializeFetchBatch(int correlationId, IEnumerable<I
                         })
                     })
                 };
-                return fetchRequest.Serialize(_requestPool.Reserve(), correlationId, _clientId, null, GetApiVersion(RequestType.BatchedFetch, _compatibility));
+                return fetchRequest.Serialize(_requestPool.Reserve(), correlationId, _clientId, null, Basics.GetApiVersion(RequestType.BatchedFetch, _compatibility));
             }
 
             public ReusableMemoryStream SerializeOffsetBatch(int correlationId, IEnumerable<IGrouping<string, OffsetMessage>> batch)
@@ -375,7 +375,7 @@ public ReusableMemoryStream SerializeOffsetBatch(int correlationId, IEnumerable<
                         })
                     })
                 };
-                return offsetRequest.Serialize(_requestPool.Reserve(), correlationId, _clientId, null, GetApiVersion(RequestType.BatchedOffset, _compatibility));
+                return offsetRequest.Serialize(_requestPool.Reserve(), correlationId, _clientId, null, Basics.GetApiVersion(RequestType.BatchedOffset, _compatibility));
             }
 
             public ReusableMemoryStream SerializeRequest(int correlationId, ISerializableRequest request, Basics.ApiVersion version)
@@ -1938,80 +1938,7 @@ private void OnNoMoreRequestSlot()
 
         private Basics.ApiVersion GetApiVersion(RequestType type)
         {
-            return GetApiVersion(type, _configuration.Compatibility);
-        }
-
-        private static Basics.ApiVersion GetApiVersion(RequestType type, Compatibility compVersion)
-        {
-            switch (type)
-            {
-                case RequestType.Metadata:
-                    return Basics.ApiVersion.Ignored;
-
-                case RequestType.BatchedProduce:
-                    switch (compVersion)
-                    {
-                        case Compatibility.V0_8_2:
-                            return Basics.ApiVersion.V0;
-                        case Compatibility.V0_10_1:
-                            return Basics.ApiVersion.V2;
-                        case Compatibility.V0_11_0:
-                            return Basics.ApiVersion.V3;
-                        default:
-                            throw new ArgumentOutOfRangeException();
-                    }
-                case RequestType.BatchedFetch:
-                    switch (compVersion)
-                    {
-                        case Compatibility.V0_8_2:
-                            return Basics.ApiVersion.V0;
-                        case Compatibility.V0_10_1:
-                            return Basics.ApiVersion.V2;
-                        case Compatibility.V0_11_0:
-                            return Basics.ApiVersion.V4;
-                        default:
-                            throw new ArgumentOutOfRangeException();
-                    }
-
-                case RequestType.BatchedOffset:
-                case RequestType.SingleOffset:
-                    switch (compVersion)
-                    {
-                        case Compatibility.V0_8_2:
-                            return Basics.ApiVersion.V0;
-                        default:
-                            return Basics.ApiVersion.V1;
-                    }
-
-                case RequestType.GroupCoordinator:
-                    return Basics.ApiVersion.V0;
-
-                case RequestType.Heartbeat:
-                    return Basics.ApiVersion.V0;
-
-                case RequestType.LeaveGroup:
-                    return Basics.ApiVersion.V0;
-
-                case RequestType.SyncConsumerGroup:
-                    return Basics.ApiVersion.V0;
-
-                case RequestType.JoinConsumerGroup:
-                    switch (compVersion)
-                    {
-                        case Compatibility.V0_8_2:
-                            return Basics.ApiVersion.V0;
-                        default:
-                            return Basics.ApiVersion.V1;
-                    }
-
-                case RequestType.OffsetCommit:
-                    return Basics.ApiVersion.V2;
-
-                case RequestType.OffsetFetch:
-                    return Basics.ApiVersion.V1;
-                default:
-                    throw new ArgumentOutOfRangeException(nameof(type), type, null);
-            }
+            return Basics.GetApiVersion(type, _configuration.Compatibility);
         }
     }
 }
diff --git a/kafka-sharp/kafka-sharp/Protocol/Basics.cs b/kafka-sharp/kafka-sharp/Protocol/Basics.cs
index d181478..35b80f4 100644
--- a/kafka-sharp/kafka-sharp/Protocol/Basics.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/Basics.cs
@@ -5,6 +5,7 @@
 using System.Collections.Generic;
 using System.IO;
 using System.IO.Compression;
+using Kafka.Cluster;
 using Kafka.Common;
 using Kafka.Public;
 #if !NETSTANDARD1_3
@@ -561,5 +562,78 @@ internal static void Uncompress(ReusableMemoryStream uncompressed, byte[] body,
         }
 
         #endregion
+
+        public static ApiVersion GetApiVersion(Node.RequestType type, Compatibility compVersion)
+        {
+            switch (type)
+            {
+                case Node.RequestType.Metadata:
+                    return ApiVersion.Ignored;
+
+                case Node.RequestType.BatchedProduce:
+                    switch (compVersion)
+                    {
+                        case Compatibility.V0_8_2:
+                            return ApiVersion.V0;
+                        case Compatibility.V0_10_1:
+                            return ApiVersion.V2;
+                        case Compatibility.V0_11_0:
+                            return ApiVersion.V3;
+                        default:
+                            throw new ArgumentOutOfRangeException();
+                    }
+                case Node.RequestType.BatchedFetch:
+                    switch (compVersion)
+                    {
+                        case Compatibility.V0_8_2:
+                            return ApiVersion.V0;
+                        case Compatibility.V0_10_1:
+                            return ApiVersion.V2;
+                        case Compatibility.V0_11_0:
+                            return ApiVersion.V4;
+                        default:
+                            throw new ArgumentOutOfRangeException();
+                    }
+
+                case Node.RequestType.BatchedOffset:
+                case Node.RequestType.SingleOffset:
+                    switch (compVersion)
+                    {
+                        case Compatibility.V0_8_2:
+                            return ApiVersion.V0;
+                        default:
+                            return ApiVersion.V1;
+                    }
+
+                case Node.RequestType.GroupCoordinator:
+                    return ApiVersion.V0;
+
+                case Node.RequestType.Heartbeat:
+                    return ApiVersion.V0;
+
+                case Node.RequestType.LeaveGroup:
+                    return ApiVersion.V0;
+
+                case Node.RequestType.SyncConsumerGroup:
+                    return ApiVersion.V0;
+
+                case Node.RequestType.JoinConsumerGroup:
+                    switch (compVersion)
+                    {
+                        case Compatibility.V0_8_2:
+                            return ApiVersion.V0;
+                        default:
+                            return ApiVersion.V1;
+                    }
+
+                case Node.RequestType.OffsetCommit:
+                    return ApiVersion.V2;
+
+                case Node.RequestType.OffsetFetch:
+                    return ApiVersion.V1;
+                default:
+                    throw new ArgumentOutOfRangeException(nameof(type), type, null);
+            }
+        }
     }
 }

From 1b9500b51f3ca2581f4682fcc47bee126533bb99 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Gr=C3=A9goire=20Verdier?= <g.verdier@criteo.com>
Date: Fri, 10 May 2019 10:50:18 +0200
Subject: [PATCH 05/15] Fix record batch serialization

- when SerializeOnProduce is on the SerializeKeyValue is now done
  correctly for records
- when SerializeOnProduce is off but the object is not sized,
  serialization is still done on produce

Change-Id: I9aa42448f9914c0b3eab95c806772981d965b1e4
---
 .../kafka-sharp.UTest/TestSerialization.cs    | 154 +++++++++++++++++-
 kafka-sharp/kafka-sharp/Protocol/Basics.cs    |  90 +++++-----
 kafka-sharp/kafka-sharp/Protocol/Message.cs   |  18 +-
 .../kafka-sharp/Protocol/ProduceRequest.cs    |  38 ++---
 .../kafka-sharp/Protocol/RecordBatch.cs       |  51 ++++--
 .../kafka-sharp/Routing/ProducerRouter.cs     |   2 +-
 6 files changed, 265 insertions(+), 88 deletions(-)

diff --git a/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs b/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs
index 2fdd9a7..adf2569 100644
--- a/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs
+++ b/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs
@@ -80,13 +80,96 @@ public void TestSerializeOneMessage_V0()
         public void TestSerializeOneMessageWithPreserializedKeyValue()
         {
             var message = new Message { Key = Key, Value = Value };
-            message.SerializeKeyValue(new ReusableMemoryStream(null), new Tuple<ISerializer, ISerializer>(null, null));
+            message.SerializeKeyValue(new ReusableMemoryStream(null), new Tuple<ISerializer, ISerializer>(null, null), Compatibility.V0_8_2);
             Assert.IsNull(message.Value);
             Assert.IsNotNull(message.SerializedKeyValue);
             TestSerializeOneMessageCommon(message);
             message.SerializedKeyValue.Dispose();
         }
 
+        [Test]
+        [TestCase(true)]
+        [TestCase(false)]
+        public void TestSerializeRecordBatch(bool withPreserializing)
+        {
+            var messages = new[] {new Message {Key = Key, Value = Value}, new Message {Key = Key, Value = Value}};
+            if (withPreserializing)
+                PreserializeMessage(messages, Compatibility.V0_11_0);
+
+            using (var stream = new ReusableMemoryStream(null))
+            {
+                var set = new PartitionData
+                {
+                    Partition = 42,
+                    CompressionCodec = CompressionCodec.None,
+                    Messages = messages,
+                };
+                set.Serialize(stream, SerializationConfig.ByteArraySerializers, Basics.ApiVersion.V3);
+                stream.Position = 0;
+
+                CheckRecordBatch(stream, set, new [] {(Key, Value), (Key, Value)});
+
+                Assert.AreEqual(stream.Length, stream.Position);
+            }
+        }
+
+        [Test]
+        [TestCase(true)]
+        [TestCase(false)]
+        public void TestSerializeRecordBatchMemorySerializable(bool withPreserializing)
+        {
+            var val = new SimpleSerializable(Encoding.UTF8.GetBytes("toto"));
+            var messages = new[] {new Message {Key = Key, Value = val}, new Message {Key = Key, Value = val}};
+            if (withPreserializing)
+                PreserializeMessage(messages, Compatibility.V0_11_0);
+
+            var pool = new Pool<ReusableMemoryStream>(() => new ReusableMemoryStream(Pool), (m, b) => { m.SetLength(0); });
+            using (var stream = new ReusableMemoryStream(pool))
+            {
+                var set = new PartitionData
+                {
+                    Partition = 42,
+                    CompressionCodec = CompressionCodec.None,
+                    Messages = messages,
+                };
+                set.Serialize(stream, SerializationConfig.ByteArraySerializers, Basics.ApiVersion.V3);
+                stream.Position = 0;
+
+                CheckRecordBatch(stream, set, new [] {(Key, val.Data), (Key, val.Data)});
+
+                Assert.AreEqual(stream.Length, stream.Position);
+            }
+        }
+
+        [Test]
+        [TestCase(true)]
+        [TestCase(false)]
+        public void TestSerializeRecordBatchSizedMemorySerializable(bool withPreserializing)
+        {
+            var val = new SimpleSizedSerializable(Encoding.UTF8.GetBytes("toto"));
+            var messages = new[] {new Message {Key = Key, Value = val}, new Message {Key = Key, Value = val}};
+
+            if (withPreserializing)
+                PreserializeMessage(messages, Compatibility.V0_11_0);
+
+            using (var stream = new ReusableMemoryStream(null))
+            {
+                var set = new PartitionData
+                {
+                    Partition = 42,
+                    CompressionCodec = CompressionCodec.None,
+                    Messages = messages,
+                };
+                set.Serialize(stream, SerializationConfig.ByteArraySerializers, Basics.ApiVersion.V3);
+                stream.Position = 0;
+
+                CheckRecordBatch(stream, set, new [] {(Key, val.Data), (Key, val.Data)});
+
+                Assert.AreEqual(stream.Length, stream.Position);
+            }
+        }
+
+
         class SimpleSerializable : IMemorySerializable
         {
             public byte[] Data;
@@ -126,7 +209,7 @@ public void TestSerializeOneMessageIMemorySerializable()
         public void TestSerializeOneMessageIMemorySerializableWithPreserializedKeyValue()
         {
             var message = new Message { Key = new SimpleSerializable(Key), Value = new SimpleSerializable(Value) };
-            message.SerializeKeyValue(new ReusableMemoryStream(null), new Tuple<ISerializer, ISerializer>(null, null));
+            message.SerializeKeyValue(new ReusableMemoryStream(null), new Tuple<ISerializer, ISerializer>(null, null), Compatibility.V0_8_2);
             Assert.IsNull(message.Value);
             Assert.IsNotNull(message.SerializedKeyValue);
             TestSerializeOneMessageCommon(message);
@@ -223,7 +306,7 @@ public void TestSerializeOneEmptySerializableMessage(MessageVersion messageversi
         public void TestSerializeOneEmptyMessageWithPreserializedKeyValue(MessageVersion messageVersion)
         {
             var message = new Message { Key = new byte[0], Value = new byte[0] };
-            message.SerializeKeyValue(new ReusableMemoryStream(null), new Tuple<ISerializer, ISerializer>(null, null));
+            message.SerializeKeyValue(new ReusableMemoryStream(null), new Tuple<ISerializer, ISerializer>(null, null), Compatibility.V0_8_2);
             Assert.IsNull(message.Value);
             Assert.IsNotNull(message.SerializedKeyValue);
             TestSerializeOneEmptyMessageCommon(message, messageVersion);
@@ -251,7 +334,7 @@ public void TestSerializeMessageWithInvalidPreserializedKeyValue()
         {
             var logger = new TestLogger();
             var message = new Message { Key = Key, Value = Value };
-            message.SerializeKeyValue(new ReusableMemoryStream(null, logger), new Tuple<ISerializer, ISerializer>(null, null));
+            message.SerializeKeyValue(new ReusableMemoryStream(null, logger), new Tuple<ISerializer, ISerializer>(null, null), Compatibility.V0_8_2);
 
             // Simulate a buffer with no data
             message.SerializedKeyValue = new ReusableMemoryStream(null, logger);
@@ -2457,5 +2540,68 @@ public long SerializedSize(object input)
                 return Value.Length;
             }
         }
+
+        private static void CheckRecordBatch(ReusableMemoryStream stream, PartitionData expected, (byte[] Key, byte[] Value)[] expectedKeyValue)
+        {
+            Assert.AreEqual(expected.Partition, BigEndianConverter.ReadInt32(stream), "partition");
+            Assert.AreEqual(stream.Length - 2 * sizeof(int), BigEndianConverter.ReadInt32(stream), "size");
+            Assert.AreEqual(0, BigEndianConverter.ReadInt64(stream), "baseOffset");
+            Assert.AreEqual(stream.Length - (3 * sizeof(int) + sizeof(long)), BigEndianConverter.ReadInt32(stream), "batchLength");
+            Assert.AreEqual(-1, BigEndianConverter.ReadInt32(stream), "partitionLeaderEpoch");
+            Assert.AreEqual(2, stream.ReadByte(), "magic");
+
+            int crcRead = BigEndianConverter.ReadInt32(stream);
+            int crcComputed = (int)Crc32.ComputeCastagnoli(stream, stream.Position, stream.Length - stream.Position);
+            Assert.AreEqual(crcComputed, crcRead, "crc");
+
+            short attributes = BigEndianConverter.ReadInt16(stream);
+            Assert.AreEqual(0, attributes, "attributes");
+
+            Message[] messages = expected.Messages.ToArray();
+
+            Assert.AreEqual(messages.Length - 1, BigEndianConverter.ReadInt32(stream), "lastOffsetDelta");
+            Assert.AreEqual(0, BigEndianConverter.ReadInt64(stream), "firstTimestamp");
+            Assert.AreEqual(0, BigEndianConverter.ReadInt64(stream), "maxTimestamp");
+            Assert.AreEqual(-1, BigEndianConverter.ReadInt64(stream), "producerId");
+            Assert.AreEqual(-1, BigEndianConverter.ReadInt16(stream), "producerEpoch");
+            Assert.AreEqual(-1, BigEndianConverter.ReadInt32(stream), "baseSequence");
+            Assert.AreEqual(messages.Length, BigEndianConverter.ReadInt32(stream), "recordsCount");
+
+            int offset = 0;
+            foreach (var message in messages)
+            {
+                byte[] expectedKey = expectedKeyValue[offset].Key;
+                byte[] expectedVal = expectedKeyValue[offset].Value;
+
+                // key + val + 1 byte + 5 varint
+                Assert.AreEqual(expectedKey.Length + expectedVal.Length + 6 * sizeof(byte), VarIntConverter.ReadInt64(stream), "length");
+                Assert.AreEqual(0, stream.ReadByte(), "attributes");
+                Assert.AreEqual(0, VarIntConverter.ReadInt64(stream), "timestampDelta");
+                Assert.AreEqual(offset, VarIntConverter.ReadInt64(stream), "offsetDelta");
+
+                Assert.AreEqual(expectedKey.Length, VarIntConverter.ReadInt64(stream), "keyLength");
+                var key = new byte[expectedKey.Length];
+                stream.Read(key, 0, key.Length);
+                Assert.AreEqual(key, expectedKey, "key");
+
+                Assert.AreEqual(expectedVal.Length, VarIntConverter.ReadInt64(stream), "valueLength");
+                var value = new byte[expectedVal.Length];
+                stream.Read(value, 0, value.Length);
+                Assert.AreEqual(value, expectedVal, "value");
+
+                Assert.AreEqual(message.Headers?.Count ?? 0, VarIntConverter.ReadInt64(stream), "headersCount");
+
+                offset += 1;
+            }
+        }
+
+        private void PreserializeMessage(Message[] messages, Compatibility compat)
+        {
+            for (int i = 0; i < messages.Length; i++)
+            {
+                messages[i].SerializeKeyValue(new ReusableMemoryStream(Pool),
+                    new Tuple<ISerializer, ISerializer>(null, null), compat);
+            }
+        }
     }
 }
\ No newline at end of file
diff --git a/kafka-sharp/kafka-sharp/Protocol/Basics.cs b/kafka-sharp/kafka-sharp/Protocol/Basics.cs
index 35b80f4..17b510e 100644
--- a/kafka-sharp/kafka-sharp/Protocol/Basics.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/Basics.cs
@@ -362,60 +362,60 @@ public static object DeserializeByteArray(ReusableMemoryStream stream, IDeserial
         /// <param name="serializer">serializer to use if object is not serializable</param>
         public static void WriteObject(ReusableMemoryStream stream, object o, ISerializer serializer)
         {
-            if (o is byte[] asBytes)
-            {
-                SerializeBytesWithVarIntSize(stream, asBytes);
-            }
-            else if (o is string asString)
-            {
-                SerializeStringWithVarIntSize(stream, asString);
-            }
-            else
+            switch (o)
             {
-                if (o == null)
-                {
+                case null:
                     stream.Write(MinusOneVarInt, 0, MinusOneVarInt.Length);
-                    return;
-                }
-
-                var expectedPosition = -1L;
-                if (o is ISizedMemorySerializable asSizedSerializable)
-                {
-                    var expectedSize = asSizedSerializable.SerializedSize();
-                    VarIntConverter.Write(stream, expectedSize);
-                    expectedPosition = stream.Position + expectedSize;
-                    asSizedSerializable.Serialize(stream);
-                }
-                else if (serializer is ISizableSerializer sizableSerializer)
-                {
-                    var expectedSize = sizableSerializer.SerializedSize(o);
-                    VarIntConverter.Write(stream, expectedSize);
-                    expectedPosition = stream.Position + expectedSize;
-                    sizableSerializer.Serialize(o, stream);
-                }
-                else
+                    break;
+                case byte[] asBytes:
+                    SerializeBytesWithVarIntSize(stream, asBytes);
+                    break;
+                case string asString:
+                    SerializeStringWithVarIntSize(stream, asString);
+                    break;
+                default:
                 {
-                    // If we can not know the size of the serialized object in advance, we need to use an intermediate buffer.
-                    using (var buffer = stream.Pool.Reserve())
+                    long expectedPosition;
+                    if (o is ISizedMemorySerializable asSizedSerializable)
                     {
-                        if (o is IMemorySerializable asSerializable)
-                        {
-                            asSerializable.Serialize(buffer);
-                        }
-                        else
+                        long expectedSize = asSizedSerializable.SerializedSize();
+                        VarIntConverter.Write(stream, expectedSize);
+                        expectedPosition = stream.Position + expectedSize;
+                        asSizedSerializable.Serialize(stream);
+                    }
+                    else if (serializer is ISizableSerializer sizableSerializer)
+                    {
+                        long expectedSize = sizableSerializer.SerializedSize(o);
+                        VarIntConverter.Write(stream, expectedSize);
+                        expectedPosition = stream.Position + expectedSize;
+                        sizableSerializer.Serialize(o, stream);
+                    }
+                    else
+                    {
+                        // If we can not know the size of the serialized object in advance, we need to use an intermediate buffer.
+                        using (var buffer = stream.Pool.Reserve())
                         {
-                            serializer.Serialize(o, buffer);
+                            if (o is IMemorySerializable asSerializable)
+                            {
+                                asSerializable.Serialize(buffer);
+                            }
+                            else
+                            {
+                                serializer.Serialize(o, buffer);
+                            }
+                            WriteObject(stream, buffer, null);
                         }
-                        WriteObject(stream, buffer, null);
+
+                        return;
                     }
 
-                    return;
-                }
+                    if (expectedPosition != stream.Position)
+                    {
+                        throw new Exception(
+                            "SerializedSize() returned a different value than the size of the serialized object written");
+                    }
 
-                if (expectedPosition != stream.Position)
-                {
-                    throw new Exception(
-                        "SerializedSize() returned a different value than the size of the serialized object written");
+                    break;
                 }
             }
         }
diff --git a/kafka-sharp/kafka-sharp/Protocol/Message.cs b/kafka-sharp/kafka-sharp/Protocol/Message.cs
index 0db2f9d..6bea6bf 100644
--- a/kafka-sharp/kafka-sharp/Protocol/Message.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/Message.cs
@@ -3,6 +3,7 @@
 
 using System;
 using System.Collections.Generic;
+using Kafka.Cluster;
 using Kafka.Common;
 using Kafka.Public;
 
@@ -27,10 +28,17 @@ internal struct Message
 
         private const int MinimumValidSizeForSerializedKeyValue = 2 * 4; // At least 4 bytes for key size and 4 bytes for value size
 
-        public void SerializeKeyValue(ReusableMemoryStream target, Tuple<ISerializer, ISerializer> serializers)
+        public void SerializeKeyValue(ReusableMemoryStream target, Tuple<ISerializer, ISerializer> serializers, Compatibility compatibility)
         {
             SerializedKeyValue = target;
-            DoSerializeKeyValue(SerializedKeyValue, serializers);
+            if (Basics.GetApiVersion(Node.RequestType.BatchedProduce, compatibility) >= Basics.ApiVersion.V3)
+            {
+                DoSerializeKeyValueAsRecord(SerializedKeyValue, serializers);
+            }
+            else
+            {
+                DoSerializeKeyValue(SerializedKeyValue, serializers);
+            }
             Value = null;
         }
 
@@ -95,6 +103,12 @@ private void HandleInvalidSerializedKeyValue(ReusableMemoryStream stream)
             stream.Write(Basics.MinusOne32, 0, 4);
         }
 
+        private void DoSerializeKeyValueAsRecord(ReusableMemoryStream stream, Tuple<ISerializer, ISerializer> serializers)
+        {
+            Basics.WriteObject(stream, Key, serializers.Item1);
+            Basics.WriteObject(stream, Value, serializers.Item2);
+        }
+
         private void DoSerializeKeyValue(ReusableMemoryStream stream, Tuple<ISerializer, ISerializer> serializers)
         {
             if (Key == null)
diff --git a/kafka-sharp/kafka-sharp/Protocol/ProduceRequest.cs b/kafka-sharp/kafka-sharp/Protocol/ProduceRequest.cs
index e0bb9a2..80f98a9 100644
--- a/kafka-sharp/kafka-sharp/Protocol/ProduceRequest.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/ProduceRequest.cs
@@ -82,35 +82,35 @@ private void SerializeRecordBatch(ReusableMemoryStream stream, ISerializer keySe
             var batch = new RecordBatch
             {
                 CompressionCodec = CompressionCodec,
-                Records = Messages.Select(message => new Record
+                Records = Messages.Select(message =>
                 {
-                    Key = EnsureSizedSerializable(message.Key, keySerializer),
-                    Value = EnsureSizedSerializable(message.Value, valueSerializer),
-                    Headers = message.Headers,
-                    Timestamp = message.TimeStamp,
-                    // If the serializer is not compatible, we already resolved this
-                    // previously, so it's ok if the cast returns null
-                    KeySerializer = keySerializer as ISizableSerializer,
-                    ValueSerializer = valueSerializer as ISizableSerializer
+                    SerializeMessageIfNotSized(ref message, keySerializer, valueSerializer, stream.Pool);
+                    return new Record
+                    {
+                        Key = message.Key,
+                        Value = message.Value,
+                        Headers = message.Headers,
+                        Timestamp = message.TimeStamp,
+                        // If the serializer is not compatible, we already resolved this
+                        // previously, so it's ok if the cast returns null
+                        KeySerializer = keySerializer as ISizableSerializer,
+                        ValueSerializer = valueSerializer as ISizableSerializer,
+                        SerializedKeyValue = message.SerializedKeyValue,
+                    };
                 }),
-
             };
 
             Basics.WriteSizeInBytes(stream, batch.Serialize);
         }
 
-        private static object EnsureSizedSerializable(object o, ISerializer serializer)
+        private static void SerializeMessageIfNotSized(ref Message msg, ISerializer keySerializer, ISerializer valueSerializer, Pool<ReusableMemoryStream> pool)
         {
-            if (Basics.SizeOfSerializedObject(o, serializer as ISizableSerializer) == Basics.UnknownSize)
+            if (msg.SerializedKeyValue == null
+                && (Basics.SizeOfSerializedObject(msg.Key, keySerializer as ISizableSerializer) == Basics.UnknownSize
+                    || Basics.SizeOfSerializedObject(msg.Value, valueSerializer as ISizableSerializer) == Basics.UnknownSize))
             {
-                using (ReusableMemoryStream buffer = new ReusableMemoryStream(null))
-                {
-                    serializer.Serialize(o, buffer);
-                    return buffer.GetBuffer();
-                }
+                msg.SerializeKeyValue(pool.Reserve(), new Serializers(keySerializer, valueSerializer), Compatibility.V0_11_0);
             }
-
-            return o;
         }
 
         private void SerializeMessageSet(ReusableMemoryStream stream, Serializers serializers, MessageVersion version)
diff --git a/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs b/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs
index 79029bc..004ff9b 100644
--- a/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs
@@ -1,7 +1,6 @@
 // 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
 
-using System;
 using System.Collections.Generic;
 using Kafka.Common;
 using Kafka.Public;
@@ -301,9 +300,12 @@ internal struct Record
 
         public ICollection<KafkaRecordHeader> Headers;
 
+        public ReusableMemoryStream SerializedKeyValue;
+
         public ReusableMemoryStream Serialize(ReusableMemoryStream target, long baseTimestamp, long offsetDelta)
         {
-            var timestampDelta = Timestamp - baseTimestamp;
+            long timestampDelta = Timestamp - baseTimestamp;
+
             VarIntConverter.Write(target, SizeOfBodyInBytes(offsetDelta, timestampDelta));
 
             // Record attributes are always null.
@@ -312,8 +314,15 @@ public ReusableMemoryStream Serialize(ReusableMemoryStream target, long baseTime
             VarIntConverter.Write(target, timestampDelta);
             VarIntConverter.Write(target, offsetDelta);
 
-            Basics.WriteObject(target, Key, KeySerializer);
-            Basics.WriteObject(target, Value, ValueSerializer);
+            if (SerializedKeyValue == null)
+            {
+                Basics.WriteObject(target, Key, KeySerializer);
+                Basics.WriteObject(target, Value, ValueSerializer);
+            }
+            else
+            {
+                target.Write(SerializedKeyValue.GetBuffer(), 0, (int)SerializedKeyValue.Length);
+            }
 
             if (Headers == null)
             {
@@ -342,24 +351,32 @@ public long SizeOfBodyInBytes(long offsetDelta, long timestampDelta)
             size += VarIntConverter.SizeOfVarInt(offsetDelta);
             size += VarIntConverter.SizeOfVarInt(timestampDelta);
 
-            if (Key == null)
-            {
-                size += Basics.MinusOneVarInt.Length;
-            }
-            else
-            {
-                var keySize = Basics.SizeOfSerializedObject(Key, KeySerializer);
-                size += keySize + VarIntConverter.SizeOfVarInt(keySize);
-            }
 
-            if (Value == null)
+            if (SerializedKeyValue == null)
             {
-                size += Basics.MinusOneVarInt.Length;
+                if (Key == null)
+                {
+                    size += Basics.MinusOneVarInt.Length;
+                }
+                else
+                {
+                    var keySize = Basics.SizeOfSerializedObject(Key, KeySerializer);
+                    size += keySize + VarIntConverter.SizeOfVarInt(keySize);
+                }
+
+                if (Value == null)
+                {
+                    size += Basics.MinusOneVarInt.Length;
+                }
+                else
+                {
+                    var valueSize = Basics.SizeOfSerializedObject(Value, ValueSerializer);
+                    size += valueSize + VarIntConverter.SizeOfVarInt(valueSize);
+                }
             }
             else
             {
-                var valueSize = Basics.SizeOfSerializedObject(Value, ValueSerializer);
-                size += valueSize + VarIntConverter.SizeOfVarInt(valueSize);
+                size += SerializedKeyValue.Length;
             }
 
             if (Headers == null)
diff --git a/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs b/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs
index 069ea9d..5677915 100644
--- a/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs
+++ b/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs
@@ -340,7 +340,7 @@ public void Route(string topic, Message message, int partition, DateTime expirat
             if (_configuration.SerializationConfig.SerializeOnProduce)
             {
                 var serializers = _configuration.SerializationConfig.GetSerializersForTopic(topic);
-                message.SerializeKeyValue(_pool.Reserve(), serializers);
+                message.SerializeKeyValue(_pool.Reserve(), serializers, _configuration.Compatibility);
             }
             Route(ProduceMessage.New(topic, partition, message, expirationDate));
         }

From d794f339d4c058aeef4593c65847a47dcafdea17 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Gr=C3=A9goire=20Verdier?= <g.verdier@criteo.com>
Date: Fri, 10 May 2019 14:03:58 +0200
Subject: [PATCH 06/15] Rename WriteSizeInBytes -> WriteWithSize

Change-Id: Ia5e0975778586c7c0b46583a0fb24b81295e9fa8
---
 kafka-sharp/kafka-sharp.UTest/TestSerialization.cs        | 2 +-
 kafka-sharp/kafka-sharp/Protocol/Basics.cs                | 6 +++---
 kafka-sharp/kafka-sharp/Protocol/ConsumerGroupRequests.cs | 4 ++--
 .../kafka-sharp/Protocol/ConsumerGroupResponses.cs        | 4 ++--
 kafka-sharp/kafka-sharp/Protocol/FetchResponse.cs         | 6 +++---
 kafka-sharp/kafka-sharp/Protocol/Message.cs               | 2 +-
 kafka-sharp/kafka-sharp/Protocol/ProduceRequest.cs        | 8 ++++----
 7 files changed, 16 insertions(+), 16 deletions(-)

diff --git a/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs b/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs
index adf2569..d72160e 100644
--- a/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs
+++ b/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs
@@ -2168,7 +2168,7 @@ public void TestDeserializeSyncConsumerGroupResponse()
             using (var serialized = new ReusableMemoryStream(null))
             {
                 BigEndianConverter.Write(serialized, (short) ErrorCode.NoError);
-                Basics.WriteSizeInBytes(serialized, stream =>
+                Basics.WriteWithSize(serialized, stream =>
                 {
                     BigEndianConverter.Write(stream, (short) 48);
                     BigEndianConverter.Write(stream, 1);
diff --git a/kafka-sharp/kafka-sharp/Protocol/Basics.cs b/kafka-sharp/kafka-sharp/Protocol/Basics.cs
index 17b510e..9fb0183 100644
--- a/kafka-sharp/kafka-sharp/Protocol/Basics.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/Basics.cs
@@ -208,21 +208,21 @@ static void WriteHeader(ReusableMemoryStream stream, long initPos)
             stream.Position = pos;
         }
 
-        public static void WriteSizeInBytes(ReusableMemoryStream stream, Action<ReusableMemoryStream> write)
+        public static void WriteWithSize(ReusableMemoryStream stream, Action<ReusableMemoryStream> write)
         {
             var initPos = ReserveHeader(stream);
             write(stream);
             WriteHeader(stream, initPos);
         }
 
-        public static void WriteSizeInBytes<T>(ReusableMemoryStream stream, T t, Action<ReusableMemoryStream, T> write)
+        public static void WriteWithSize<T>(ReusableMemoryStream stream, T t, Action<ReusableMemoryStream, T> write)
         {
             var initPos = ReserveHeader(stream);
             write(stream, t);
             WriteHeader(stream, initPos);
         }
 
-        public static void WriteSizeInBytes<T, U>(ReusableMemoryStream stream, T t, U u, Action<ReusableMemoryStream, T, U> write)
+        public static void WriteWithSize<T, U>(ReusableMemoryStream stream, T t, U u, Action<ReusableMemoryStream, T, U> write)
         {
             var initPos = ReserveHeader(stream);
             write(stream, t, u);
diff --git a/kafka-sharp/kafka-sharp/Protocol/ConsumerGroupRequests.cs b/kafka-sharp/kafka-sharp/Protocol/ConsumerGroupRequests.cs
index 73f02a8..8acd482 100644
--- a/kafka-sharp/kafka-sharp/Protocol/ConsumerGroupRequests.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/ConsumerGroupRequests.cs
@@ -52,7 +52,7 @@ public void Serialize(ReusableMemoryStream stream, object _, Basics.ApiVersion _
         {
             Basics.SerializeString(stream, ProtocolName);
             var pm = ProtocolMetadata;
-            Basics.WriteSizeInBytes(stream, s => pm.Serialize(s, null, Basics.ApiVersion.Ignored));
+            Basics.WriteWithSize(stream, s => pm.Serialize(s, null, Basics.ApiVersion.Ignored));
         }
 
         #region Deserialization (for test)
@@ -157,7 +157,7 @@ public void Serialize(ReusableMemoryStream stream, object _, Basics.ApiVersion _
         {
             Basics.SerializeString(stream, MemberId);
             var ma = MemberAssignment;
-            Basics.WriteSizeInBytes(stream, s => ma.Serialize(s, null, Basics.ApiVersion.Ignored));
+            Basics.WriteWithSize(stream, s => ma.Serialize(s, null, Basics.ApiVersion.Ignored));
         }
 
         public void Deserialize(ReusableMemoryStream stream, object _, Basics.ApiVersion __)
diff --git a/kafka-sharp/kafka-sharp/Protocol/ConsumerGroupResponses.cs b/kafka-sharp/kafka-sharp/Protocol/ConsumerGroupResponses.cs
index 45cc84d..f067f73 100644
--- a/kafka-sharp/kafka-sharp/Protocol/ConsumerGroupResponses.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/ConsumerGroupResponses.cs
@@ -22,7 +22,7 @@ public void Serialize(ReusableMemoryStream stream, object _, Basics.ApiVersion _
         {
             Basics.SerializeString(stream, MemberId);
             var pm = Metadata;
-            Basics.WriteSizeInBytes(stream, s => pm.Serialize(s, null, Basics.ApiVersion.Ignored));
+            Basics.WriteWithSize(stream, s => pm.Serialize(s, null, Basics.ApiVersion.Ignored));
         }
 
         #endregion
@@ -89,7 +89,7 @@ public void Serialize(ReusableMemoryStream stream, object _, Basics.ApiVersion _
         {
             BigEndianConverter.Write(stream, (short) ErrorCode);
             var ma = MemberAssignment;
-            Basics.WriteSizeInBytes(stream, s => ma.Serialize(s, null, Basics.ApiVersion.Ignored));
+            Basics.WriteWithSize(stream, s => ma.Serialize(s, null, Basics.ApiVersion.Ignored));
         }
 
         public void Deserialize(ReusableMemoryStream stream, object _, Basics.ApiVersion __)
diff --git a/kafka-sharp/kafka-sharp/Protocol/FetchResponse.cs b/kafka-sharp/kafka-sharp/Protocol/FetchResponse.cs
index 3350977..2b20ca0 100644
--- a/kafka-sharp/kafka-sharp/Protocol/FetchResponse.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/FetchResponse.cs
@@ -328,15 +328,15 @@ public void Serialize(ReusableMemoryStream stream, object extra, Basics.ApiVersi
                         ValueSerializer = null
                     }),
                 };
-                Basics.WriteSizeInBytes(stream, batch.Serialize);
+                Basics.WriteWithSize(stream, batch.Serialize);
                 return;
             }
-            Basics.WriteSizeInBytes(stream, Messages, (s, l) =>
+            Basics.WriteWithSize(stream, Messages, (s, l) =>
             {
                 foreach (var m in l)
                 {
                     BigEndianConverter.Write(s, m.Offset);
-                    Basics.WriteSizeInBytes(s, m.Message,
+                    Basics.WriteWithSize(s, m.Message,
                         (st, msg) =>
                             msg.Serialize(st, CompressionCodec.None, extra as Serializers,
                                 version == Basics.ApiVersion.V2 ? MessageVersion.V1 : MessageVersion.V0));
diff --git a/kafka-sharp/kafka-sharp/Protocol/Message.cs b/kafka-sharp/kafka-sharp/Protocol/Message.cs
index 6bea6bf..96907f2 100644
--- a/kafka-sharp/kafka-sharp/Protocol/Message.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/Message.cs
@@ -141,7 +141,7 @@ private static void SerializeObject(ReusableMemoryStream stream, ISerializer ser
             }
             else
             {
-                Basics.WriteSizeInBytes(stream, theValue, serializer, SerializerWrite);
+                Basics.WriteWithSize(stream, theValue, serializer, SerializerWrite);
             }
         }
 
diff --git a/kafka-sharp/kafka-sharp/Protocol/ProduceRequest.cs b/kafka-sharp/kafka-sharp/Protocol/ProduceRequest.cs
index 80f98a9..37850da 100644
--- a/kafka-sharp/kafka-sharp/Protocol/ProduceRequest.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/ProduceRequest.cs
@@ -100,7 +100,7 @@ private void SerializeRecordBatch(ReusableMemoryStream stream, ISerializer keySe
                 }),
             };
 
-            Basics.WriteSizeInBytes(stream, batch.Serialize);
+            Basics.WriteWithSize(stream, batch.Serialize);
         }
 
         private static void SerializeMessageIfNotSized(ref Message msg, ISerializer keySerializer, ISerializer valueSerializer, Pool<ReusableMemoryStream> pool)
@@ -115,7 +115,7 @@ private static void SerializeMessageIfNotSized(ref Message msg, ISerializer keyS
 
         private void SerializeMessageSet(ReusableMemoryStream stream, Serializers serializers, MessageVersion version)
         {
-            Basics.WriteSizeInBytes(stream, Messages,
+            Basics.WriteWithSize(stream, Messages,
                 new SerializationInfo
                 {
                     Serializers = serializers,
@@ -137,7 +137,7 @@ private static void SerializeMessagesUncompressed(ReusableMemoryStream stream, I
                 // When message format is V0, brokers will rewrite the offsets anyway
                 // so we use the same scheme in all cases.
                 BigEndianConverter.Write(stream, offset++);
-                Basics.WriteSizeInBytes(stream, message,
+                Basics.WriteWithSize(stream, message,
                     new SerializationInfo
                     {
                         CompressionCodec = CompressionCodec.None,
@@ -173,7 +173,7 @@ private static void _SerializeMessages(ReusableMemoryStream stream, IEnumerable<
                             Value = compressed,
                             TimeStamp = Timestamp.Now
                         };
-                        Basics.WriteSizeInBytes(stream, m,
+                        Basics.WriteWithSize(stream, m,
                             new SerializationInfo
                             {
                                 Serializers = SerializationConfig.ByteArraySerializers,

From a8cbbc702a290c1beb9f06ba1a123eee417f34df Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Gr=C3=A9goire=20Verdier?= <g.verdier@criteo.com>
Date: Fri, 10 May 2019 14:11:22 +0200
Subject: [PATCH 07/15] Rename VarIntConvert.Read{Int64, Int32, ...} to
 ReadAs{Int64, Int32, ...}

ReadInt32 implies there is an Int32 in the stream but it actually
reads a varint and stores it in an Int32

Change-Id: I375566a8173b1a74ea4b9906a14a9ccf29a53d21
---
 .../kafka-sharp.UTest/TestSerialization.cs    | 36 +++++++++----------
 .../kafka-sharp.UTest/TestVarIntConverter.cs  | 18 +++++-----
 .../kafka-sharp/Common/VarIntConverter.cs     | 18 +++++-----
 kafka-sharp/kafka-sharp/Protocol/Basics.cs    |  4 +--
 .../kafka-sharp/Protocol/RecordBatch.cs       | 12 +++----
 5 files changed, 44 insertions(+), 44 deletions(-)

diff --git a/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs b/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs
index d72160e..c303874 100644
--- a/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs
+++ b/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs
@@ -930,17 +930,17 @@ public void TestSerializeProduceRequest_V3(string transactionalId, CompressionCo
                 var records = Decompress(serialized, codec);
 
                 // First record
-                var firstRecordLength = VarIntConverter.ReadInt64(records);
+                var firstRecordLength = VarIntConverter.ReadAsInt64(records);
                 var startOfFirstRecord = records.Position;
                 var recordsCount = 0L;
                 Assert.AreEqual(0, records.ReadByte()); // attributes
-                Assert.AreEqual(0, VarIntConverter.ReadInt64(records)); // TimeStampDelta
-                Assert.AreEqual(0, VarIntConverter.ReadInt64(records)); // offsetDelta
-                Assert.AreEqual(-1, VarIntConverter.ReadInt64(records)); // keyLength
+                Assert.AreEqual(0, VarIntConverter.ReadAsInt64(records)); // TimeStampDelta
+                Assert.AreEqual(0, VarIntConverter.ReadAsInt64(records)); // offsetDelta
+                Assert.AreEqual(-1, VarIntConverter.ReadAsInt64(records)); // keyLength
                 // Since the keyLength is -1, we do not read the key
-                Assert.AreEqual(Value.Length, VarIntConverter.ReadInt64(records)); //value length
+                Assert.AreEqual(Value.Length, VarIntConverter.ReadAsInt64(records)); //value length
                 Assert.AreEqual(TheValue, deserializer.Deserialize(records, Value.Length));
-                recordsCount = VarIntConverter.ReadInt64(records);
+                recordsCount = VarIntConverter.ReadAsInt64(records);
                 Assert.AreEqual(2L, recordsCount); // Headers count
                 for (var i = 0; i < recordsCount; ++i)
                 {
@@ -951,16 +951,16 @@ public void TestSerializeProduceRequest_V3(string transactionalId, CompressionCo
                 Assert.AreEqual(firstRecordLength, endOfFirstRecord - startOfFirstRecord);
 
                 // Second record
-                var secondRecordLength = VarIntConverter.ReadInt64(records);
+                var secondRecordLength = VarIntConverter.ReadAsInt64(records);
                 var startOfSecondRecord = records.Position;
                 Assert.AreEqual(0, records.ReadByte()); // attributes
-                Assert.AreEqual(0, VarIntConverter.ReadInt64(records)); // TimeStampDelta
-                Assert.AreEqual(1, VarIntConverter.ReadInt64(records)); // offsetDelta
-                Assert.AreEqual(-1, VarIntConverter.ReadInt64(records)); // keyLength
+                Assert.AreEqual(0, VarIntConverter.ReadAsInt64(records)); // TimeStampDelta
+                Assert.AreEqual(1, VarIntConverter.ReadAsInt64(records)); // offsetDelta
+                Assert.AreEqual(-1, VarIntConverter.ReadAsInt64(records)); // keyLength
                 // Since the keyLength is -1, we do not read the key
-                Assert.AreEqual(Value.Length, VarIntConverter.ReadInt64(records)); //value length
+                Assert.AreEqual(Value.Length, VarIntConverter.ReadAsInt64(records)); //value length
                 Assert.AreEqual(TheValue, deserializer.Deserialize(records, Value.Length));
-                recordsCount = VarIntConverter.ReadInt64(records);
+                recordsCount = VarIntConverter.ReadAsInt64(records);
                 Assert.AreEqual(2L, recordsCount); // Headers count
                 for (var i = 0; i < recordsCount; ++i)
                 {
@@ -2574,22 +2574,22 @@ private static void CheckRecordBatch(ReusableMemoryStream stream, PartitionData
                 byte[] expectedVal = expectedKeyValue[offset].Value;
 
                 // key + val + 1 byte + 5 varint
-                Assert.AreEqual(expectedKey.Length + expectedVal.Length + 6 * sizeof(byte), VarIntConverter.ReadInt64(stream), "length");
+                Assert.AreEqual(expectedKey.Length + expectedVal.Length + 6 * sizeof(byte), VarIntConverter.ReadAsInt64(stream), "length");
                 Assert.AreEqual(0, stream.ReadByte(), "attributes");
-                Assert.AreEqual(0, VarIntConverter.ReadInt64(stream), "timestampDelta");
-                Assert.AreEqual(offset, VarIntConverter.ReadInt64(stream), "offsetDelta");
+                Assert.AreEqual(0, VarIntConverter.ReadAsInt64(stream), "timestampDelta");
+                Assert.AreEqual(offset, VarIntConverter.ReadAsInt64(stream), "offsetDelta");
 
-                Assert.AreEqual(expectedKey.Length, VarIntConverter.ReadInt64(stream), "keyLength");
+                Assert.AreEqual(expectedKey.Length, VarIntConverter.ReadAsInt64(stream), "keyLength");
                 var key = new byte[expectedKey.Length];
                 stream.Read(key, 0, key.Length);
                 Assert.AreEqual(key, expectedKey, "key");
 
-                Assert.AreEqual(expectedVal.Length, VarIntConverter.ReadInt64(stream), "valueLength");
+                Assert.AreEqual(expectedVal.Length, VarIntConverter.ReadAsInt64(stream), "valueLength");
                 var value = new byte[expectedVal.Length];
                 stream.Read(value, 0, value.Length);
                 Assert.AreEqual(value, expectedVal, "value");
 
-                Assert.AreEqual(message.Headers?.Count ?? 0, VarIntConverter.ReadInt64(stream), "headersCount");
+                Assert.AreEqual(message.Headers?.Count ?? 0, VarIntConverter.ReadAsInt64(stream), "headersCount");
 
                 offset += 1;
             }
diff --git a/kafka-sharp/kafka-sharp.UTest/TestVarIntConverter.cs b/kafka-sharp/kafka-sharp.UTest/TestVarIntConverter.cs
index c767399..2acbd00 100644
--- a/kafka-sharp/kafka-sharp.UTest/TestVarIntConverter.cs
+++ b/kafka-sharp/kafka-sharp.UTest/TestVarIntConverter.cs
@@ -48,41 +48,41 @@ public void TestSizeOfVarInt()
         [TestCase(new byte[] { 0x80, 0x80, 0x80, 0x80, 0x80, 0x80, 0x80, 0x80, 0x80, 0x02 })]
         public void TestReadInt64Overflows(byte[] value)
         {
-            Assert.Throws<OverflowException>(() => VarIntConverter.ReadInt64(new MemoryStream(value)));
+            Assert.Throws<OverflowException>(() => VarIntConverter.ReadAsInt64(new MemoryStream(value)));
         }
 
         [Test]
         public void TestReadInt64()
         {
-            RunReadIntTestFor(VarIntConverter.ReadInt64, long.MinValue, long.MaxValue);
+            RunReadIntTestFor(VarIntConverter.ReadAsInt64, long.MinValue, long.MaxValue);
         }
 
         [Test]
         public void TestReadInt32()
         {
-            RunReadIntTestFor(VarIntConverter.ReadInt32, int.MinValue, int.MaxValue);
+            RunReadIntTestFor(VarIntConverter.ReadAsInt32, int.MinValue, int.MaxValue);
         }
 
         [Test]
         public void TestReadInt16()
         {
-            RunReadIntTestFor(VarIntConverter.ReadInt16, short.MinValue, short.MaxValue);
+            RunReadIntTestFor(VarIntConverter.ReadAsInt16, short.MinValue, short.MaxValue);
         }
 
         [Test]
         public void TestReadByte()
         {
-            RunReadIntTestFor(VarIntConverter.ReadByte, byte.MinValue, byte.MaxValue, true);
+            RunReadIntTestFor(VarIntConverter.ReadAsByte, byte.MinValue, byte.MaxValue, true);
         }
 
         [Test]
         public void TestReadBool()
         {
-            Assert.IsFalse(VarIntConverter.ReadBool(new MemoryStream(new byte[] { 0x00 })));
+            Assert.IsFalse(VarIntConverter.ReadAsBool(new MemoryStream(new byte[] { 0x00 })));
 
-            Assert.IsTrue(VarIntConverter.ReadBool(new MemoryStream(new byte[] { 0x01 })));
-            Assert.IsTrue(VarIntConverter.ReadBool(new MemoryStream(new byte[] { 0x02 })));
-            Assert.IsTrue(VarIntConverter.ReadBool(new MemoryStream(new byte[] { 0x80, 0x01 })));
+            Assert.IsTrue(VarIntConverter.ReadAsBool(new MemoryStream(new byte[] { 0x01 })));
+            Assert.IsTrue(VarIntConverter.ReadAsBool(new MemoryStream(new byte[] { 0x02 })));
+            Assert.IsTrue(VarIntConverter.ReadAsBool(new MemoryStream(new byte[] { 0x80, 0x01 })));
         }
 
         private void RunReadIntTestFor<TInteger>(Func<MemoryStream, TInteger> readFunc, TInteger minValue,
diff --git a/kafka-sharp/kafka-sharp/Common/VarIntConverter.cs b/kafka-sharp/kafka-sharp/Common/VarIntConverter.cs
index 959547b..db4d970 100644
--- a/kafka-sharp/kafka-sharp/Common/VarIntConverter.cs
+++ b/kafka-sharp/kafka-sharp/Common/VarIntConverter.cs
@@ -85,7 +85,7 @@ public static void Write(MemoryStream stream, bool i)
             Write(stream, i ? 1L : 0L);
         }
 
-        public static long ReadInt64(MemoryStream stream)
+        public static long ReadAsInt64(MemoryStream stream)
         {
             ulong asZigZag = 0L; // Result value
             int i = 0; // Number of bits written
@@ -122,24 +122,24 @@ public static long ReadInt64(MemoryStream stream)
             return (long)(asZigZag >> 1);
         }
 
-        public static int ReadInt32(MemoryStream stream)
+        public static int ReadAsInt32(MemoryStream stream)
         {
-            return checked((int)ReadInt64(stream));
+            return checked((int)ReadAsInt64(stream));
         }
 
-        public static short ReadInt16(MemoryStream stream)
+        public static short ReadAsInt16(MemoryStream stream)
         {
-            return checked((short)ReadInt64(stream));
+            return checked((short)ReadAsInt64(stream));
         }
 
-        public static byte ReadByte(MemoryStream stream)
+        public static byte ReadAsByte(MemoryStream stream)
         {
-            return checked((byte)ReadInt64(stream));
+            return checked((byte)ReadAsInt64(stream));
         }
 
-        public static bool ReadBool(MemoryStream stream)
+        public static bool ReadAsBool(MemoryStream stream)
         {
-            return ReadInt64(stream) != 0;
+            return ReadAsInt64(stream) != 0;
         }
     }
 }
diff --git a/kafka-sharp/kafka-sharp/Protocol/Basics.cs b/kafka-sharp/kafka-sharp/Protocol/Basics.cs
index 9fb0183..b25fce1 100644
--- a/kafka-sharp/kafka-sharp/Protocol/Basics.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/Basics.cs
@@ -110,7 +110,7 @@ public static void SerializeBytes(ReusableMemoryStream stream, byte[] b)
 
         public static byte[] DeserializeBytesWithVarIntSize(ReusableMemoryStream stream)
         {
-            var len = VarIntConverter.ReadInt32(stream);
+            var len = VarIntConverter.ReadAsInt32(stream);
             if (len == -1)
                 return null;
 
@@ -163,7 +163,7 @@ public static void SerializeString(ReusableMemoryStream stream, string s)
 
         public static string DeserializeStringWithVarIntSize(ReusableMemoryStream stream)
         {
-            var len = VarIntConverter.ReadInt32(stream);
+            var len = VarIntConverter.ReadAsInt32(stream);
             // per contract, null string is represented with -1 len.
             if (len == -1)
                 return null;
diff --git a/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs b/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs
index 004ff9b..cd951a7 100644
--- a/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs
@@ -251,7 +251,7 @@ public IEnumerable<Record> DeserializeRecordsUncompressed(ReusableMemoryStream i
         {
             for (var i = 0; i < numberOfRecords; i++)
             {
-                var length = VarIntConverter.ReadInt32(input);
+                var length = VarIntConverter.ReadAsInt32(input);
                 if (input.Length - input.Position < length)
                 {
                     throw new ProtocolException(
@@ -259,15 +259,15 @@ public IEnumerable<Record> DeserializeRecordsUncompressed(ReusableMemoryStream i
                 }
 
                 var attributes = input.ReadByte(); // ignored for now
-                var timeStampDelta = VarIntConverter.ReadInt64(input);
-                var offsetDelta = VarIntConverter.ReadInt32(input);
+                var timeStampDelta = VarIntConverter.ReadAsInt64(input);
+                var offsetDelta = VarIntConverter.ReadAsInt32(input);
 
-                var keyLength = VarIntConverter.ReadInt32(input);
+                var keyLength = VarIntConverter.ReadAsInt32(input);
                 var key = keyLength == -1 ? null : deserializers.Item1.Deserialize(input, keyLength);
-                var valueLength = VarIntConverter.ReadInt32(input);
+                var valueLength = VarIntConverter.ReadAsInt32(input);
                 var value = valueLength == -1 ? null : deserializers.Item1.Deserialize(input, valueLength);
 
-                var headersCount = VarIntConverter.ReadInt32(input);
+                var headersCount = VarIntConverter.ReadAsInt32(input);
                 var headers = new List<KafkaRecordHeader>(headersCount);
                 for (var j = 0; j < headersCount; j++)
                 {

From 61f6219d0cbfaff7b2f0ce3e65b838f1f06f3dce Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Gr=C3=A9goire=20Verdier?= <g.verdier@criteo.com>
Date: Fri, 10 May 2019 14:31:45 +0200
Subject: [PATCH 08/15] Make RecordBatch.Deserialize static

This change is to replace this pattern

var recordBatch = new RecordBatch();
recordBatch.Deserialize(stream, deserializers, endOfAllBatches);

with

var recordBatch = RecordBatch.Deserialize(stream, deserializers, endOfAllBatches);

The big diff is because I moved static method at the top of the class

Change-Id: I381a0145bc147495fa4145fd17e99d8bf14968b5
---
 .../kafka-sharp/Protocol/FetchResponse.cs     |   3 +-
 .../kafka-sharp/Protocol/RecordBatch.cs       | 224 +++++++++---------
 2 files changed, 115 insertions(+), 112 deletions(-)

diff --git a/kafka-sharp/kafka-sharp/Protocol/FetchResponse.cs b/kafka-sharp/kafka-sharp/Protocol/FetchResponse.cs
index 2b20ca0..5f9f4a4 100644
--- a/kafka-sharp/kafka-sharp/Protocol/FetchResponse.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/FetchResponse.cs
@@ -152,7 +152,6 @@ internal static List<ResponseMessage> DeserializeRecordBatch(ReusableMemoryStrea
                 return DeserializeMessageSet(stream, deserializers);
             }
             // Now we know what we received is a proper recordBatch
-            var recordBatch = new RecordBatch();
             var size = BigEndianConverter.ReadInt32(stream);
             var endOfAllBatches = stream.Position + size;
             if (stream.Length < endOfAllBatches)
@@ -162,7 +161,7 @@ internal static List<ResponseMessage> DeserializeRecordBatch(ReusableMemoryStrea
             }
             while (stream.Position < endOfAllBatches)
             {
-                recordBatch.Deserialize(stream, deserializers, endOfAllBatches);
+                var recordBatch = RecordBatch.Deserialize(stream, deserializers, endOfAllBatches);
                 list.AddRange(recordBatch.Records.Select(record => new ResponseMessage
                 {
                     Message = new Message { Key = record.Key, Value = record.Value, TimeStamp = record.Timestamp },
diff --git a/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs b/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs
index cd951a7..99cd72f 100644
--- a/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs
@@ -51,6 +51,120 @@ internal class RecordBatch
         private const int BytesNecessaryToGetLength = 8 // baseOffset
             + 4; // batchLength
 
+        // Remark: it might be the case that brokers are authorized to send us partial record batch.
+        // So if the protocol exceptions in this method are triggered, you might want to investigate and remove
+        // them altogether.
+        public static RecordBatch Deserialize(ReusableMemoryStream input, Deserializers deserializers, long endOfAllBatches)
+        {
+            var recordBatch = new RecordBatch();
+
+            if (input.Position + BytesNecessaryToGetLength > endOfAllBatches)
+            {
+                throw new ProtocolException(
+                    $"Trying to read a batch record at {input.Position} and the end of all batches is {endOfAllBatches}."
+                    + $" There is not enough bytes remaining to even read the first fields...");
+            }
+            recordBatch.BaseOffset = BigEndianConverter.ReadInt64(input);
+            var batchLength = BigEndianConverter.ReadInt32(input);
+            var endOfBatch = input.Position + batchLength;
+            if (endOfAllBatches < endOfBatch)
+            {
+                throw new ProtocolException(
+                    $"The record batch says it has length that stops at {endOfBatch} but the list of all batches stop at {endOfAllBatches}.");
+            }
+            recordBatch.PartitionLeaderEpoch = BigEndianConverter.ReadInt32(input);
+            var magic = input.ReadByte();
+            // Current magic value is 2
+            if ((uint) magic != 2)
+            {
+                throw new UnsupportedMagicByteVersion((byte) magic, "2");
+            }
+
+            var crc = (uint) BigEndianConverter.ReadInt32(input);
+            var afterCrcPosition = input.Position; // The crc is calculated starting from this position
+            Crc32.CheckCrcCastagnoli((int)crc, input, afterCrcPosition, endOfBatch - afterCrcPosition);
+
+            var attributes = BigEndianConverter.ReadInt16(input);
+
+            recordBatch.CompressionCodec = (CompressionCodec) (attributes & CompressionCodecMask);
+            recordBatch.IsTransactional = (attributes & TransactionalFlagMask) != 0;
+            recordBatch.IsControl = (attributes & ControlFlagMask) != 0;
+            recordBatch.TimestampType = (attributes & TimestampTypeMask) > 0
+                ? TimestampType.LogAppendTime
+                : TimestampType.CreateTime;
+
+            var lastOffsetDelta = BigEndianConverter.ReadInt32(input);
+
+            var firstTimestamp = BigEndianConverter.ReadInt64(input);
+            var maxTimestamp = BigEndianConverter.ReadInt64(input);
+
+            recordBatch.ProducerId = BigEndianConverter.ReadInt64(input);
+            recordBatch.ProducerEpoch = BigEndianConverter.ReadInt16(input);
+            recordBatch.BaseSequence = BigEndianConverter.ReadInt32(input);
+
+            var numberOfRecords = BigEndianConverter.ReadInt32(input);
+            recordBatch.Records = DeserializeRecords(recordBatch, input, numberOfRecords, endOfBatch, firstTimestamp, deserializers);
+
+            return recordBatch;
+        }
+
+        public static IEnumerable<Record> DeserializeRecords(RecordBatch recordBatch, ReusableMemoryStream input, int numberOfRecords, long endOfBatch,
+            long firstTimeStamp, Deserializers deserializers)
+        {
+            if (recordBatch.CompressionCodec == CompressionCodec.None)
+            {
+                return DeserializeRecordsUncompressed(recordBatch, input, numberOfRecords, endOfBatch, firstTimeStamp,
+                    deserializers);
+            }
+            using (var uncompressedStream = input.Pool.Reserve())
+            {
+                Basics.Uncompress(uncompressedStream, input.GetBuffer(), (int) input.Position,
+                    (int) (endOfBatch - input.Position), recordBatch.CompressionCodec);
+                input.Position = endOfBatch;
+                return new List<Record>(DeserializeRecordsUncompressed(recordBatch, uncompressedStream, numberOfRecords, endOfBatch, firstTimeStamp,
+                    deserializers)); // We use a list here to force iteration to take place, so that we can release uncompressedStream
+            }
+        }
+
+        public static IEnumerable<Record> DeserializeRecordsUncompressed(RecordBatch recordBatch, ReusableMemoryStream input, int numberOfRecords, long endOfBatch,
+            long firstTimeStamp, Deserializers deserializers)
+        {
+            for (var i = 0; i < numberOfRecords; i++)
+            {
+                var length = VarIntConverter.ReadAsInt32(input);
+                if (input.Length - input.Position < length)
+                {
+                    throw new ProtocolException(
+                        $"Record said it was of length {length}, but actually only {input.Length - input.Position} bytes remain");
+                }
+
+                var attributes = input.ReadByte(); // ignored for now
+                var timeStampDelta = VarIntConverter.ReadAsInt64(input);
+                var offsetDelta = VarIntConverter.ReadAsInt32(input);
+
+                var keyLength = VarIntConverter.ReadAsInt32(input);
+                var key = keyLength == -1 ? null : deserializers.Item1.Deserialize(input, keyLength);
+                var valueLength = VarIntConverter.ReadAsInt32(input);
+                var value = valueLength == -1 ? null : deserializers.Item1.Deserialize(input, valueLength);
+
+                var headersCount = VarIntConverter.ReadAsInt32(input);
+                var headers = new List<KafkaRecordHeader>(headersCount);
+                for (var j = 0; j < headersCount; j++)
+                {
+                    headers.Add(Record.DeserializeHeader(input));
+                }
+
+                yield return new Record
+                {
+                    Headers = headers,
+                    Key = key,
+                    Timestamp = firstTimeStamp + timeStampDelta,
+                    Value = value,
+                    Offset = recordBatch.BaseOffset + offsetDelta
+                };
+            }
+        }
+
         public void Serialize(ReusableMemoryStream target)
         {
             uint crc = 0;
@@ -174,116 +288,6 @@ public void Serialize(ReusableMemoryStream target)
                 }
             }
         }
-
-        // Remark: it might be the case that brokers are authorized to send us partial record batch.
-        // So if the protocol exceptions in this method are triggered, you might want to investigate and remove
-        // them altogether.
-        public void Deserialize(ReusableMemoryStream input, Deserializers deserializers, long endOfAllBatches)
-        {
-            if (input.Position + BytesNecessaryToGetLength > endOfAllBatches)
-            {
-                throw new ProtocolException(
-                    $"Trying to read a batch record at {input.Position} and the end of all batches is {endOfAllBatches}."
-                    + $" There is not enough bytes remaining to even read the first fields...");
-            }
-            BaseOffset = BigEndianConverter.ReadInt64(input);
-            var batchLength = BigEndianConverter.ReadInt32(input);
-            var endOfBatch = input.Position + batchLength;
-            if (endOfAllBatches < endOfBatch)
-            {
-                throw new ProtocolException(
-                    $"The record batch says it has length that stops at {endOfBatch} but the list of all batches stop at {endOfAllBatches}.");
-            }
-            PartitionLeaderEpoch = BigEndianConverter.ReadInt32(input);
-            var magic = input.ReadByte();
-            // Current magic value is 2
-            if ((uint) magic != 2)
-            {
-                throw new UnsupportedMagicByteVersion((byte) magic, "2");
-            }
-
-            var crc = (uint) BigEndianConverter.ReadInt32(input);
-            var afterCrcPosition = input.Position; // The crc is calculated starting from this position
-            Crc32.CheckCrcCastagnoli((int)crc, input, afterCrcPosition, endOfBatch - afterCrcPosition);
-
-            var attributes = BigEndianConverter.ReadInt16(input);
-
-            CompressionCodec = (CompressionCodec) (attributes & CompressionCodecMask);
-            IsTransactional = (attributes & TransactionalFlagMask) != 0;
-            IsControl = (attributes & ControlFlagMask) != 0;
-            TimestampType = (attributes & TimestampTypeMask) > 0
-                ? TimestampType.LogAppendTime
-                : TimestampType.CreateTime;
-
-            var lastOffsetDelta = BigEndianConverter.ReadInt32(input);
-
-            var firstTimestamp = BigEndianConverter.ReadInt64(input);
-            var maxTimestamp = BigEndianConverter.ReadInt64(input);
-
-            ProducerId = BigEndianConverter.ReadInt64(input);
-            ProducerEpoch = BigEndianConverter.ReadInt16(input);
-            BaseSequence = BigEndianConverter.ReadInt32(input);
-
-            var numberOfRecords = BigEndianConverter.ReadInt32(input);
-            Records = DeserializeRecords(input, numberOfRecords, endOfBatch, firstTimestamp, deserializers);
-        }
-
-        public IEnumerable<Record> DeserializeRecords(ReusableMemoryStream input, int numberOfRecords, long endOfBatch,
-            long firstTimeStamp, Deserializers deserializers)
-        {
-            if (CompressionCodec == CompressionCodec.None)
-            {
-                return DeserializeRecordsUncompressed(input, numberOfRecords, endOfBatch, firstTimeStamp,
-                    deserializers);
-            }
-            using (var uncompressedStream = input.Pool.Reserve())
-            {
-                Basics.Uncompress(uncompressedStream, input.GetBuffer(), (int) input.Position,
-                    (int) (endOfBatch - input.Position), CompressionCodec);
-                input.Position = endOfBatch;
-                return new List<Record>(DeserializeRecordsUncompressed(uncompressedStream, numberOfRecords, endOfBatch, firstTimeStamp,
-                    deserializers)); // We use a list here to force iteration to take place, so that we can release uncompressedStream
-            }
-        }
-
-        public IEnumerable<Record> DeserializeRecordsUncompressed(ReusableMemoryStream input, int numberOfRecords, long endOfBatch,
-            long firstTimeStamp, Deserializers deserializers)
-        {
-            for (var i = 0; i < numberOfRecords; i++)
-            {
-                var length = VarIntConverter.ReadAsInt32(input);
-                if (input.Length - input.Position < length)
-                {
-                    throw new ProtocolException(
-                        $"Record said it was of length {length}, but actually only {input.Length - input.Position} bytes remain");
-                }
-
-                var attributes = input.ReadByte(); // ignored for now
-                var timeStampDelta = VarIntConverter.ReadAsInt64(input);
-                var offsetDelta = VarIntConverter.ReadAsInt32(input);
-
-                var keyLength = VarIntConverter.ReadAsInt32(input);
-                var key = keyLength == -1 ? null : deserializers.Item1.Deserialize(input, keyLength);
-                var valueLength = VarIntConverter.ReadAsInt32(input);
-                var value = valueLength == -1 ? null : deserializers.Item1.Deserialize(input, valueLength);
-
-                var headersCount = VarIntConverter.ReadAsInt32(input);
-                var headers = new List<KafkaRecordHeader>(headersCount);
-                for (var j = 0; j < headersCount; j++)
-                {
-                    headers.Add(Record.DeserializeHeader(input));
-                }
-
-                yield return new Record
-                {
-                    Headers = headers,
-                    Key = key,
-                    Timestamp = firstTimeStamp + timeStampDelta,
-                    Value = value,
-                    Offset = BaseOffset + offsetDelta
-                };
-            }
-        }
     }
 
     internal struct Record

From bf4884d10a50b3e0edd3d9c1301fea6cb62de864 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Gr=C3=A9goire=20Verdier?= <g.verdier@criteo.com>
Date: Fri, 10 May 2019 14:36:25 +0200
Subject: [PATCH 09/15] Remove useless temporary variable

Change-Id: I84fd77041218e4dd9fb1acadc6dd406103f982b5
---
 kafka-sharp/kafka-sharp/Protocol/Message.cs | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/kafka-sharp/kafka-sharp/Protocol/Message.cs b/kafka-sharp/kafka-sharp/Protocol/Message.cs
index 96907f2..3eef4ed 100644
--- a/kafka-sharp/kafka-sharp/Protocol/Message.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/Message.cs
@@ -135,9 +135,8 @@ private static void SerializeObject(ReusableMemoryStream stream, ISerializer ser
             // byte[] are just copied
             if (theValue is byte[] bytes)
             {
-                byte[] array = bytes;
-                BigEndianConverter.Write(stream, array.Length);
-                stream.Write(array, 0, array.Length);
+                BigEndianConverter.Write(stream, bytes.Length);
+                stream.Write(bytes, 0, bytes.Length);
             }
             else
             {

From ca4261c39620af046906653496b8981a47a07e87 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Gr=C3=A9goire=20Verdier?= <g.verdier@criteo.com>
Date: Fri, 10 May 2019 14:45:15 +0200
Subject: [PATCH 10/15] Declare RecordBatch.Serialize variables as locally as
 possible

Change-Id: I43d6650ee70bbff8661b6406ab5f4086ca0381f2
---
 .../kafka-sharp/Protocol/RecordBatch.cs       | 23 ++++++++-----------
 1 file changed, 9 insertions(+), 14 deletions(-)

diff --git a/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs b/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs
index 99cd72f..527bf6b 100644
--- a/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs
@@ -167,13 +167,9 @@ public static IEnumerable<Record> DeserializeRecordsUncompressed(RecordBatch rec
 
         public void Serialize(ReusableMemoryStream target)
         {
-            uint crc = 0;
-            int batchLength = -1, lastOffsetDelta = -1;
-            long firstTimestamp = -1L, maxTimestamp = -1L;
-
             BigEndianConverter.Write(target, BaseOffset);
             var batchLengthPosition = target.Position;
-            BigEndianConverter.Write(target, batchLength); // placeholder for batchLength: int32
+            BigEndianConverter.Write(target, -1); // placeholder for batchLength: int32
             // The value of batchLength is the number of bytes required to read after the
             // batchLength field, we retain the current buffer position to compute it later.
             var afterBatchLengthPosition = target.Position;
@@ -183,7 +179,7 @@ public void Serialize(ReusableMemoryStream target)
             target.WriteByte(2);
 
             var crcPosition = target.Position;
-            BigEndianConverter.Write(target, (int)crc); // placeholder for CRC: int32
+            BigEndianConverter.Write(target, 0); // placeholder for CRC: int32
             var afterCrcPosition = target.Position;
 
             short attributes = 0x0;
@@ -199,29 +195,28 @@ public void Serialize(ReusableMemoryStream target)
             BigEndianConverter.Write(target, attributes);
 
             var lastOffsetDeltaPosition = target.Position;
-            BigEndianConverter.Write(target, lastOffsetDelta); // placeholder for LastOffsetDelta: int32
+            BigEndianConverter.Write(target, -1); // placeholder for LastOffsetDelta: int32
 
             var firstTimestampPosition = target.Position;
-            BigEndianConverter.Write(target, firstTimestamp); // placeholder for FirstTimestamp: int64
+            BigEndianConverter.Write(target, -1L); // placeholder for FirstTimestamp: int64
 
             var maxTimestampPosition = target.Position;
-            BigEndianConverter.Write(target, maxTimestamp); // placeholder for MaxTimestamp: int64
+            BigEndianConverter.Write(target, -1L); // placeholder for MaxTimestamp: int64
 
             BigEndianConverter.Write(target, ProducerId);
             BigEndianConverter.Write(target, ProducerEpoch);
             BigEndianConverter.Write(target, BaseSequence); // BaseSequence: int32
 
             var recordsCountPosition = target.Position;
-            BigEndianConverter.Write(target, lastOffsetDelta);  // RecordsCount: int32
+            BigEndianConverter.Write(target, -1);  // RecordsCount: int32
 
-            (lastOffsetDelta, firstTimestamp, maxTimestamp) = CompressionCodec == CompressionCodec.None
+            (int lastOffsetDelta, long firstTimestamp, long maxTimestamp) = CompressionCodec == CompressionCodec.None
                 ? SerializeRecords(target)
                 : SerializeRecordsWithCompression(target, CompressionCodec);
 
-
             var endPosition = target.Position;
 
-            batchLength = (int)(target.Position - afterBatchLengthPosition);
+            int batchLength = (int)(target.Position - afterBatchLengthPosition);
             target.Position = batchLengthPosition;
             BigEndianConverter.Write(target, batchLength);
 
@@ -238,7 +233,7 @@ public void Serialize(ReusableMemoryStream target)
             BigEndianConverter.Write(target, lastOffsetDelta + 1); // actual records count
 
             // Crc is computed on the object from right after the CRC to the end of the buffer
-            crc = Crc32.ComputeCastagnoli(target, afterCrcPosition, endPosition - afterCrcPosition);
+            uint crc = Crc32.ComputeCastagnoli(target, afterCrcPosition, endPosition - afterCrcPosition);
             target.Position = crcPosition;
             BigEndianConverter.Write(target, (int)crc);
 

From 3864c90aedb9d687c69af9e195f539cef8c09dae Mon Sep 17 00:00:00 2001
From: Jules Bovet <j.bovet@criteo.com>
Date: Thu, 4 Jul 2019 11:36:55 +0200
Subject: [PATCH 11/15] Handle brokers sending partial messages in
 FetchResponse

From the official documentation: "As an optimization the server is allowed
to return a partial message at the end of the message set. Clients should
handle this case.".

We then discard the message, the read offset not being updated for this message,
meaning we will get it in full in the next FetchRequest.

Change-Id: I0dae2e93641c7a48544d5a3b02287bd10319ab58
---
 .../kafka-sharp.UTest/TestSerialization.cs    | 75 +++++++++++++++++++
 .../kafka-sharp/Protocol/FetchResponse.cs     |  6 ++
 .../kafka-sharp/Protocol/RecordBatch.cs       | 13 ++--
 3 files changed, 89 insertions(+), 5 deletions(-)

diff --git a/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs b/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs
index c303874..39fedb9 100644
--- a/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs
+++ b/kafka-sharp/kafka-sharp.UTest/TestSerialization.cs
@@ -1617,6 +1617,81 @@ public void TestDeserializeFetchResponse_V5()
             }
         }
 
+        [Test]
+        public void TestDeserializeFetchResponse_PartialMessageEndOfMessageSet()
+        {
+            var response = new FetchResponse
+            {
+                ThrottleTime = 47,
+                FetchPartitionResponse =
+                    new CommonResponse<FetchPartitionResponse>
+                    {
+                        TopicsResponse =
+                            new[]
+                            {
+                                new TopicData<FetchPartitionResponse>
+                                {
+                                    TopicName = "The_Wire",
+                                    PartitionsData =
+                                        new[]
+                                        {
+                                            new FetchPartitionResponse
+                                            {
+                                                ErrorCode = ErrorCode.NoError,
+                                                HighWatermarkOffset = 714,
+                                                Partition = 999999,
+                                                LastStableOffset = 404,
+                                                LogStartOffset = 200,
+                                                AbortedTransactions = new[]
+                                                {
+                                                    new AbortedTransaction {FirstOffset = 3, ProducerId = 4},
+                                                },
+                                                Messages =
+                                                    new List<ResponseMessage>
+                                                    {
+                                                        new ResponseMessage
+                                                        {
+                                                            Offset = 0,
+                                                            Message = new Message {Key = Key, Value = Value}
+                                                        },
+                                                        new ResponseMessage
+                                                        {
+                                                            Offset = 1337,
+                                                            Message = new Message {Key = Key, Value = Value}
+                                                        },
+                                                    }
+                                            }
+                                        }
+                                }
+                            }
+                    }
+            };
+
+            using (var serialized = new ReusableMemoryStream(null))
+            {
+                var config = new SerializationConfig();
+                BigEndianConverter.Write(serialized, response.ThrottleTime);
+                Basics.WriteArray(serialized, response.FetchPartitionResponse.TopicsResponse, config,
+                    Basics.ApiVersion.V5);
+
+                // Expected position of where the size of the RecordBatch will be stored
+                // We change this size to something bigger than the size of the buffer to simulate a partial message
+                // being sent by the broker (= stream length < stream position + message size)
+                serialized.Position = 84;
+                var spoofedSize = BitConverter.GetBytes(500);
+                Array.Reverse(spoofedSize);
+                serialized.Write(spoofedSize, 0, 4);
+                serialized.Position = 0;
+
+                // Test that the FetchResponse Deserialize logic will not throw on this and instead
+                // discard the RecordBatch
+                var f = new FetchResponse();
+                f.Deserialize(serialized, config, Basics.ApiVersion.V5);
+                var deserializedMessages = f.FetchPartitionResponse.TopicsResponse.First().PartitionsData.First().Messages;
+                Assert.AreEqual(0, deserializedMessages.Count);
+            }
+        }
+
         /* The following test are ordered to be run first because if we break one
          * it will most likely break all other tests. Thus when serialization tests
          * break we can quick check if it's due to basic stuff being broken or not
diff --git a/kafka-sharp/kafka-sharp/Protocol/FetchResponse.cs b/kafka-sharp/kafka-sharp/Protocol/FetchResponse.cs
index 5f9f4a4..78a3c79 100644
--- a/kafka-sharp/kafka-sharp/Protocol/FetchResponse.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/FetchResponse.cs
@@ -162,6 +162,12 @@ internal static List<ResponseMessage> DeserializeRecordBatch(ReusableMemoryStrea
             while (stream.Position < endOfAllBatches)
             {
                 var recordBatch = RecordBatch.Deserialize(stream, deserializers, endOfAllBatches);
+                if (recordBatch == null)
+                {
+                    // We received a partial record batch, discard it
+                    stream.Position = endOfAllBatches;
+                    break;
+                }
                 list.AddRange(recordBatch.Records.Select(record => new ResponseMessage
                 {
                     Message = new Message { Key = record.Key, Value = record.Value, TimeStamp = record.Timestamp },
diff --git a/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs b/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs
index 527bf6b..aa3c291 100644
--- a/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs
+++ b/kafka-sharp/kafka-sharp/Protocol/RecordBatch.cs
@@ -51,9 +51,12 @@ internal class RecordBatch
         private const int BytesNecessaryToGetLength = 8 // baseOffset
             + 4; // batchLength
 
-        // Remark: it might be the case that brokers are authorized to send us partial record batch.
-        // So if the protocol exceptions in this method are triggered, you might want to investigate and remove
-        // them altogether.
+        /// <remarks>
+        /// From the official protocol documentation available at
+        /// https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-FetchAPI
+        /// "As an optimization the server is allowed to return a partial message at the end of the message set. Clients should handle this case."
+        /// If the end of the RecordBatch exceeds the length of the whole response (= endOfAllBatches), we should discard this RecordBatch.
+        /// </remarks>
         public static RecordBatch Deserialize(ReusableMemoryStream input, Deserializers deserializers, long endOfAllBatches)
         {
             var recordBatch = new RecordBatch();
@@ -69,8 +72,8 @@ public static RecordBatch Deserialize(ReusableMemoryStream input, Deserializers
             var endOfBatch = input.Position + batchLength;
             if (endOfAllBatches < endOfBatch)
             {
-                throw new ProtocolException(
-                    $"The record batch says it has length that stops at {endOfBatch} but the list of all batches stop at {endOfAllBatches}.");
+                // Partial message, CRCs won't match, return here so the CRC check doesn't throw
+                return null;
             }
             recordBatch.PartitionLeaderEpoch = BigEndianConverter.ReadInt32(input);
             var magic = input.ReadByte();

From 0ed6dc8fa4ee23e6d7a1def96bd9daa214ea59c4 Mon Sep 17 00:00:00 2001
From: Jules Bovet <j.bovet@criteo.com>
Date: Fri, 5 Jul 2019 11:24:51 +0200
Subject: [PATCH 12/15] Introduce retry delay when fetching partitions in
 HandleStart fails

Currently, when trying to start consuming a topic, if the first fetch of
the available partitions fails (because the topic does not exist for example),
the driver will immediately retry. This can translate to hundreds of requests
per second, which is useless.

As this loop in HandleStart is already blocking indefinitely the ProcessMessage()
loop of the driver, introducing a retry delay does not change the behavior.

Change-Id: Ice89ab8e4cb1a41ba25fb2bec49bd1be66001a8b
---
 kafka-sharp/kafka-sharp/Routing/ConsumerRouter.cs | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/kafka-sharp/kafka-sharp/Routing/ConsumerRouter.cs b/kafka-sharp/kafka-sharp/Routing/ConsumerRouter.cs
index 795daff..eff1665 100644
--- a/kafka-sharp/kafka-sharp/Routing/ConsumerRouter.cs
+++ b/kafka-sharp/kafka-sharp/Routing/ConsumerRouter.cs
@@ -710,8 +710,10 @@ private async Task HandleStart(TopicOnOff topicOnOff)
                         {
                             // Means the cluster is kind of dead, ignore and retry, there's not much else
                             // to do anyway since we won't be able to send more fetch requests until
-                            // some node is available.
+                            // some node is available. Also wait an arbitrary amount of time before trying
+                            // again so we don't hammer the cluster.
                             // TODO: be verbose?
+                            await Task.Delay(TimeSpan.FromSeconds(2));
                         }
                     }
             }

From 73318c6b499a09193ed2088aecd8476b07b0ccd5 Mon Sep 17 00:00:00 2001
From: Jules Bovet <j.bovet@criteo.com>
Date: Fri, 19 Jul 2019 10:43:24 +0200
Subject: [PATCH 13/15] Improve dead node logic

- Add a log when a broker is removed from the nodes list because of a
  topology change
- Remove unknown node (when dead) from the routing table so it is not
  reused, which was not the case before

Change-Id: I32cb7fe19f9935b0abe088950a07f1590b7a2951
---
 kafka-sharp/kafka-sharp.UTest/Mocks.cs        | 37 +++++++++++++++----
 kafka-sharp/kafka-sharp.UTest/TestBatching.cs |  8 ++--
 kafka-sharp/kafka-sharp.UTest/TestClient.cs   |  4 ++
 kafka-sharp/kafka-sharp.UTest/TestCluster.cs  |  5 ++-
 kafka-sharp/kafka-sharp.UTest/TestRouter.cs   |  4 ++
 .../kafka-sharp.UTest/TestRoutingTable.cs     |  4 +-
 kafka-sharp/kafka-sharp/Cluster/Cluster.cs    | 21 +++++------
 kafka-sharp/kafka-sharp/Cluster/Node.cs       | 36 +++++++++++++-----
 .../kafka-sharp/Routing/ProducerRouter.cs     |  4 +-
 .../kafka-sharp/Routing/RoutingTable.cs       |  2 +-
 10 files changed, 88 insertions(+), 37 deletions(-)

diff --git a/kafka-sharp/kafka-sharp.UTest/Mocks.cs b/kafka-sharp/kafka-sharp.UTest/Mocks.cs
index 6a071ff..3ebb2ce 100644
--- a/kafka-sharp/kafka-sharp.UTest/Mocks.cs
+++ b/kafka-sharp/kafka-sharp.UTest/Mocks.cs
@@ -110,20 +110,24 @@ public static void Reset()
         }
     }
 
-    class NodeMock : INode
+    class NodeMock : INode, IEquatable<NodeMock>
     {
         private MetadataResponse _response;
-        public NodeMock() : this(new MetadataResponse()) { }
+        public NodeMock()
+            : this("node1")
+        { }
 
-        public NodeMock(MetadataResponse response)
+        public NodeMock(string name)
+            : this(new MetadataResponse(), name)
+        { }
+
+        public NodeMock(MetadataResponse response, string name)
         {
+            Name = name;
             _response = response;
         }
 
-        public string Name
-        {
-            get { return "Some node"; }
-        }
+        public string Name { get; set; }
 
         public bool Produce(ProduceMessage message)
         {
@@ -261,6 +265,25 @@ public bool Post(IBatchByTopic<OffsetMessage> batch)
         {
             throw new NotImplementedException();
         }
+
+        public override bool Equals(object obj)
+        {
+            return Equals(obj as NodeMock);
+        }
+
+        public bool Equals(NodeMock other)
+        {
+            if (ReferenceEquals(null, other))
+                return false;
+            if (ReferenceEquals(this, other))
+                return true;
+            return string.Equals(Name, other.Name);
+        }
+
+        public override int GetHashCode()
+        {
+            return Name == null ? 0 : Name.GetHashCode();
+        }
     }
 
     class ClusterMock : ICluster
diff --git a/kafka-sharp/kafka-sharp.UTest/TestBatching.cs b/kafka-sharp/kafka-sharp.UTest/TestBatching.cs
index 0b45d31..d59e5c7 100644
--- a/kafka-sharp/kafka-sharp.UTest/TestBatching.cs
+++ b/kafka-sharp/kafka-sharp.UTest/TestBatching.cs
@@ -167,8 +167,8 @@ public void TestAccumulatorByTopicByPartitionTimeElapsed()
         [TestCase(500000, 100)]
         public void TestAccumulatorByNodeByTopic(int batchSize, int time)
         {
-            INode n1 = new NodeMock();
-            INode n2 = new NodeMock();
+            INode n1 = new NodeMock("n1");
+            INode n2 = new NodeMock("n2");
             var accumulator = new AccumulatorByNodeByTopic<Tuple<string, int>>(t => t.Item1, batchSize, TimeSpan.FromMilliseconds(time));
 
             var count = new CountdownEvent(2);
@@ -198,8 +198,8 @@ public void TestAccumulatorByNodeByTopic(int batchSize, int time)
         [TestCase(500000, 100)]
         public void TestAccumulatorByNodeByTopicByPartition(int batchSize, int time)
         {
-            INode n1 = new NodeMock();
-            INode n2 = new NodeMock();
+            INode n1 = new NodeMock("n1");
+            INode n2 = new NodeMock("n2");
             var accumulator = new AccumulatorByNodeByTopicByPartition<Tuple<string, int, int>>(t => t.Item1, t => t.Item2, batchSize, TimeSpan.FromMilliseconds(time));
 
             var count = new CountdownEvent(2);
diff --git a/kafka-sharp/kafka-sharp.UTest/TestClient.cs b/kafka-sharp/kafka-sharp.UTest/TestClient.cs
index 051535e..f45e1f5 100644
--- a/kafka-sharp/kafka-sharp.UTest/TestClient.cs
+++ b/kafka-sharp/kafka-sharp.UTest/TestClient.cs
@@ -26,11 +26,15 @@ private void Init(Configuration configuration)
         {
             _node = new Mock<INode>();
             var brokerMeta = new BrokerMeta();
+            brokerMeta.Host = "brokerHost";
             _node.Setup(n => n.FetchMetadata())
                 .Returns(Task.FromResult(new MetadataResponse
                 {
                     BrokersMeta = new[] {brokerMeta}, TopicsMeta = new TopicMeta[0]
                 }));
+            _node.Setup(n => n.Equals(It.IsAny<object>())).Returns(true);
+            _node.Setup(n => n.Equals(It.IsAny<INode>())).Returns(true);
+            _node.Setup(n => n.GetHashCode()).Returns(0);
             _producer = new Mock<IProduceRouter>();
             _consumer = new Mock<IConsumeRouter>();
             var logger = new Mock<ILogger>();
diff --git a/kafka-sharp/kafka-sharp.UTest/TestCluster.cs b/kafka-sharp/kafka-sharp.UTest/TestCluster.cs
index e8a7884..2a596b0 100644
--- a/kafka-sharp/kafka-sharp.UTest/TestCluster.cs
+++ b/kafka-sharp/kafka-sharp.UTest/TestCluster.cs
@@ -65,10 +65,13 @@ public void Setup()
         private Mock<INode> GenerateNodeMock(int port)
         {
             var nodeMock = new Mock<INode>();
-            nodeMock.Setup(n => n.Name).Returns("localhost:" + port);
+            var name = "localhost:" + port;
+            nodeMock.Setup(n => n.Name).Returns(name);
             nodeMock.Setup(n => n.FetchMetadata()).Returns(Task.FromResult(TestData.TestMetadataResponse));
             nodeMock.Setup(n => n.FetchMetadata(It.IsAny<IEnumerable<string>>())).Returns(Task.FromResult(TestData.TestMetadataResponse));
             nodeMock.Setup(n => n.Stop()).Returns(Task.FromResult(true));
+            nodeMock.Setup(n => n.Equals(It.IsAny<INode>())).Returns((INode other) => name == other.Name);
+            nodeMock.Setup(n => n.GetHashCode()).Returns(name.GetHashCode);
             return nodeMock;
         }
 
diff --git a/kafka-sharp/kafka-sharp.UTest/TestRouter.cs b/kafka-sharp/kafka-sharp.UTest/TestRouter.cs
index b8a2b45..cd4ff10 100644
--- a/kafka-sharp/kafka-sharp.UTest/TestRouter.cs
+++ b/kafka-sharp/kafka-sharp.UTest/TestRouter.cs
@@ -243,7 +243,11 @@ public void TestFilterMinInSyncEnough()
         [Test]
         public void TestMessageAreSent_GlobalBatching()
         {
+            var name = "n1";
             var node = new Mock<INode>();
+            node.SetupGet(n => n.Name).Returns(name);
+            node.Setup(n => n.Equals(It.IsAny<INode>())).Returns((INode other) => name == other.Name);
+            node.Setup(n => n.GetHashCode()).Returns(name.GetHashCode);
             node.Setup(n => n.Post(It.IsAny<IBatchByTopicByPartition<ProduceMessage>>())).Returns(true);
             var cluster = new Mock<ICluster>();
             cluster.Setup(c => c.RequireNewRoutingTable())
diff --git a/kafka-sharp/kafka-sharp.UTest/TestRoutingTable.cs b/kafka-sharp/kafka-sharp.UTest/TestRoutingTable.cs
index a265d72..3f9fa1c 100644
--- a/kafka-sharp/kafka-sharp.UTest/TestRoutingTable.cs
+++ b/kafka-sharp/kafka-sharp.UTest/TestRoutingTable.cs
@@ -58,11 +58,11 @@ public void TestRoutingTableReturnsEmptyForAbsentTopic()
         [Test]
         public void TestSignalDeadNode()
         {
-            var node = new NodeMock();
+            var node = new NodeMock("n1");
             var routes = new Dictionary<string, Partition[]>
             {
                 {"test1p", new[] {new Partition {Id = 0, Leader = node}}},
-                {"test2p", new[] {new Partition {Id = 1, Leader = new NodeMock()}, new Partition {Id = 2, Leader = node}, new Partition {Id = 3, Leader = new NodeMock()}}},
+                {"test2p", new[] {new Partition {Id = 1, Leader = new NodeMock("n2")}, new Partition {Id = 2, Leader = node}, new Partition {Id = 3, Leader = new NodeMock("n3")}}},
             };
             var routingTable = new RoutingTable(routes);
 
diff --git a/kafka-sharp/kafka-sharp/Cluster/Cluster.cs b/kafka-sharp/kafka-sharp/Cluster/Cluster.cs
index 999c441..fc2d75a 100644
--- a/kafka-sharp/kafka-sharp/Cluster/Cluster.cs
+++ b/kafka-sharp/kafka-sharp/Cluster/Cluster.cs
@@ -639,23 +639,21 @@ private async Task CheckNoMoreNodes()
         private async Task ProcessDeadNode(INode deadNode)
         {
             Statistics.UpdateNodeDead(GetNodeId(deadNode));
-            BrokerMeta m;
-            if (!_nodes.TryGetValue(deadNode, out m))
-            {
-                Logger.LogWarning(string.Format("Kafka unknown node dead, the node makes itself known as: {0}.",
-                                              deadNode.Name));
-                return;
-            }
-            Logger.LogWarning(string.Format("Kafka node {0} is dead, refreshing metadata.", GetNodeName(deadNode)));
+            Logger.LogWarning(_nodes.TryGetValue(deadNode, out var brokerMeta)
+                ? $"Kafka node {GetNodeName(deadNode)} is dead, refreshing metadata."
+                : $"Kafka unknown node dead, the node makes itself known as: {deadNode.Name}.");
+
             if (_routingTable != null)
             {
                 _routingTable = new RoutingTable(_routingTable, deadNode);
                 RoutingTableChange(_routingTable);
             }
             _nodes.Remove(deadNode);
-            deadNode.Stop();
-            _nodesByHostPort.Remove(BuildKey(m.Host, m.Port));
-            _nodesById.Remove(m.Id);
+            var _ = deadNode.Stop(); // Fire and forget
+
+            if (brokerMeta == null) return;
+            _nodesByHostPort.Remove(BuildKey(brokerMeta.Host, brokerMeta.Port));
+            _nodesById.Remove(brokerMeta.Id);
             await CheckNoMoreNodes();
             RefreshMetadata();
         }
@@ -921,6 +919,7 @@ private void ResponseToTopology(MetadataResponse response)
             {
                 var node = _nodesByHostPort[host];
                 _nodesByHostPort.Remove(host);
+                Logger.LogInformation($"Removing broker '{node.Name}' from the cluster in response to new topology");
                 _nodes.Remove(node);
                 node.Stop();
             }
diff --git a/kafka-sharp/kafka-sharp/Cluster/Node.cs b/kafka-sharp/kafka-sharp/Cluster/Node.cs
index b660c68..f6b7927 100644
--- a/kafka-sharp/kafka-sharp/Cluster/Node.cs
+++ b/kafka-sharp/kafka-sharp/Cluster/Node.cs
@@ -28,7 +28,7 @@ namespace Kafka.Cluster
     /// wants them. They're responsible for maintaining a connection to a corresponding
     /// Kafka broker.
     /// </summary>
-    interface INode
+    internal interface INode
     {
         /// <summary>
         /// A name associated to this node.
@@ -245,7 +245,6 @@ Task<JoinConsumerGroupResponse> JoinConsumerGroup(string groupId, string memberI
         /// The node reached its maximum number of concurrent requests.
         /// </summary>
         event Action<INode> NoMoreRequestSlot;
-
     }
 
     /// <summary>
@@ -258,7 +257,7 @@ Task<JoinConsumerGroupResponse> JoinConsumerGroup(string groupId, string memberI
     /// State change shared between send and receive actors is kept minimal (mainly correlation ids matching).
     /// Connection setup is always handled in the Send actor.
     /// </summary>
-    sealed class Node : INode
+    internal sealed class Node : INode, IEquatable<Node>
     {
         #region Serialization / Deserialization
 
@@ -693,19 +692,19 @@ private struct Ping
 
         #region Responses
 
-        struct ResponseData
+        private struct ResponseData
         {
             public ReusableMemoryStream Data;
             public int CorrelationId;
         }
 
-        struct ResponseException
+        private struct ResponseException
         {
             public Exception Exception;
         }
 
         [StructLayout(LayoutKind.Explicit)]
-        struct ResponseValue
+        private struct ResponseValue
         {
             [FieldOffset(0)]
             public ResponseException ResponseException;
@@ -717,14 +716,14 @@ struct ResponseValue
         /// <summary>
         /// Message types for the underlying response handler actor.
         /// </summary>
-        enum ResponseType
+        private enum ResponseType
         {
             Data,
             Stop,
             Exception
         }
 
-        struct Response
+        private struct Response
         {
             public ResponseType ResponseType;
             public IConnection Connection;
@@ -749,7 +748,7 @@ struct Response
         private readonly Configuration _configuration;
         private readonly TimeoutScheduler _timeoutScheduler;
 
-        struct Pending
+        private struct Pending
         {
             public int CorrelationId;
             public Request Request;
@@ -1940,5 +1939,24 @@ private Basics.ApiVersion GetApiVersion(RequestType type)
         {
             return Basics.GetApiVersion(type, _configuration.Compatibility);
         }
+
+        public override bool Equals(object obj)
+        {
+            return Equals(obj as Node);
+        }
+
+        public bool Equals(Node other)
+        {
+            if (ReferenceEquals(null, other))
+                return false;
+            if (ReferenceEquals(this, other))
+                return true;
+            return string.Equals(Name, other.Name);
+        }
+
+        public override int GetHashCode()
+        {
+            return Name == null ? 0 : Name.GetHashCode();
+        }
     }
 }
diff --git a/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs b/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs
index 5677915..da81ddf 100644
--- a/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs
+++ b/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs
@@ -953,8 +953,8 @@ private Dictionary<int, DateTime> GetFilter(string topic)
         private void OnMessageExpired(ProduceMessage message)
         {
             _cluster.Logger.LogError(string.Format(
-                "[Producer] Not able to send message before reaching TTL for [topic: {0} / partition: {1}], message expired.",
-                message.Topic, message.RequiredPartition));
+                "[Producer] Not able to send message before reaching TTL for [topic: {0} / partition: {1} (required {2})], message expired.",
+                message.Topic, message.Partition, message.RequiredPartition));
 
             ClearMessage(message.Message, shouldClearKeyValue: false);
             MessageExpired(message.Topic, message.Message);
diff --git a/kafka-sharp/kafka-sharp/Routing/RoutingTable.cs b/kafka-sharp/kafka-sharp/Routing/RoutingTable.cs
index f2a159c..5c8077d 100644
--- a/kafka-sharp/kafka-sharp/Routing/RoutingTable.cs
+++ b/kafka-sharp/kafka-sharp/Routing/RoutingTable.cs
@@ -68,7 +68,7 @@ public RoutingTable(RoutingTable fromTable, INode deadNode)
             var tmp = new List<Partition>();
             foreach (var kv in fromTable._routes)
             {
-                tmp.AddRange(kv.Value.Where(partition => partition.Leader != deadNode));
+                tmp.AddRange(kv.Value.Where(partition => !Equals(partition.Leader, deadNode)));
                 _routes.Add(kv.Key, tmp.ToArray());
                 tmp.Clear();
             }

From 49b7ca76db859e857e55302691b14a8e6ad9b80a Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Gr=C3=A9goire=20Verdier?= <g.verdier@criteo.com>
Date: Mon, 30 Sep 2019 11:14:59 +0200
Subject: [PATCH 14/15] Format partition correctly when logging

So that logs like "...reaching TTL for [topic: toto / partition: -3]"
are replaced with "...reaching TTL for [topic: toto / partition: any]"

Change-Id: I95f2a8400abd66ebc82aaf0336eeb4f29f39888e
---
 kafka-sharp/kafka-sharp/Public/KafkaRecord.cs  | 18 ++++++++++++++++++
 .../kafka-sharp/Routing/ProducerRouter.cs      | 10 +++++-----
 2 files changed, 23 insertions(+), 5 deletions(-)

diff --git a/kafka-sharp/kafka-sharp/Public/KafkaRecord.cs b/kafka-sharp/kafka-sharp/Public/KafkaRecord.cs
index abd8abd..23ba5ae 100644
--- a/kafka-sharp/kafka-sharp/Public/KafkaRecord.cs
+++ b/kafka-sharp/kafka-sharp/Public/KafkaRecord.cs
@@ -13,6 +13,24 @@ public static class Partitions
         public const int None = -1;
         public const int All = -2;
         public const int Any = -3;
+
+        /// <summary>
+        /// Format partition in a printable way
+        /// </summary>
+        public static string Format(int partition)
+        {
+            switch (partition)
+            {
+                case None:
+                    return "none";
+                case All:
+                    return "all";
+                case Any:
+                    return "any";
+                default:
+                    return partition.ToString();
+            }
+        }
     }
 
     /// <summary>
diff --git a/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs b/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs
index da81ddf..986be6b 100644
--- a/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs
+++ b/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs
@@ -361,7 +361,7 @@ private void Route(ProduceMessage message)
                 _cluster.Logger.LogError(
                     string.Format(
                         "[Producer] Failed to route message, discarding message for [topic: {0} / partition: {1}]",
-                        message.Topic, message.Partition));
+                        message.Topic, Partitions.Format(message.Partition)));
                 OnMessageDiscarded(message);
             }
         }
@@ -779,7 +779,7 @@ private async Task HandleProduceAcknowledgement(ProduceAcknowledgement acknowled
                             _cluster.Logger.LogError(
                                 string.Format(
                                     "[Producer] Irrecoverable error, discarding message for [topic: {0} / partition: {1}]",
-                                    pm.Topic, pm.Partition));
+                                    pm.Topic, Partitions.Format(pm.Partition)));
                             OnMessageDiscarded(pm);
                         }
                         else
@@ -858,7 +858,7 @@ private void PostponeMessage(ProduceMessage produceMessage)
                 _cluster.Logger.LogError(
                     string.Format(
                         "[Producer] Too many postponed messages, discarding message for [topic: {0} / partition: {1}]",
-                        produceMessage.Topic, produceMessage.RequiredPartition));
+                        produceMessage.Topic, Partitions.Format(produceMessage.RequiredPartition)));
                 OnMessageDiscarded(produceMessage);
                 return;
             }
@@ -876,7 +876,7 @@ private void PostponeMessage(ProduceMessage produceMessage)
                 {
                     _cluster.Logger.LogError(
                         string.Format("[Producer] No node available for [topic: {0} / partition: {1}], postponing messages.",
-                            produceMessage.Topic, produceMessage.RequiredPartition));
+                            produceMessage.Topic, Partitions.Format(produceMessage.RequiredPartition)));
                 }
                 else
                 {
@@ -954,7 +954,7 @@ private void OnMessageExpired(ProduceMessage message)
         {
             _cluster.Logger.LogError(string.Format(
                 "[Producer] Not able to send message before reaching TTL for [topic: {0} / partition: {1} (required {2})], message expired.",
-                message.Topic, message.Partition, message.RequiredPartition));
+                message.Topic, Partitions.Format(message.Partition), Partitions.Format(message.RequiredPartition)));
 
             ClearMessage(message.Message, shouldClearKeyValue: false);
             MessageExpired(message.Topic, message.Message);

From 4287a2a56d82e3ffac34bdd38775bde7fc34e4cf Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Gr=C3=A9goire=20Verdier?= <g.verdier@criteo.com>
Date: Tue, 25 Feb 2020 11:21:57 +0100
Subject: [PATCH 15/15] Remove LoggerManager

Change-Id: Ibf22b2e88b79ffae73f2cfcd08914438e96ebe7d
---
 .../kafka-sharp.UTest/TestMessageKeyPartitionSelection.cs | 7 ++++---
 .../kafka-sharp/Public/PartitionSelectionConfig.cs        | 5 +++--
 .../PartitionSelection/MessageKeyPartitionSelection.cs    | 8 +++++---
 kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs         | 3 ++-
 4 files changed, 14 insertions(+), 9 deletions(-)

diff --git a/kafka-sharp/kafka-sharp.UTest/TestMessageKeyPartitionSelection.cs b/kafka-sharp/kafka-sharp.UTest/TestMessageKeyPartitionSelection.cs
index 2beaa4a..7f32c86 100644
--- a/kafka-sharp/kafka-sharp.UTest/TestMessageKeyPartitionSelection.cs
+++ b/kafka-sharp/kafka-sharp.UTest/TestMessageKeyPartitionSelection.cs
@@ -5,6 +5,7 @@
 using Kafka.Public;
 using Kafka.Routing;
 using Kafka.Routing.PartitionSelection;
+using Moq;
 using NUnit.Framework;
 
 namespace tests_kafka_sharp
@@ -24,7 +25,7 @@ public void Test_MessageKeyPartitionSelection_Is_Consistent()
                 new Partition {Id = 1, Leader = nodeMock},
                 new Partition {Id = 2, Leader = nodeMock},
             };
-            var partitionStrategy = new MessageKeyPartitionSelection(Serializer, RoundRobinPartitionSelection);
+            var partitionStrategy = new MessageKeyPartitionSelection(Serializer, RoundRobinPartitionSelection, Mock.Of<ILogger>());
             var partitioner = new PartitionSelector(partitionStrategy);
             var message1 = ProduceMessage.New(string.Empty, Partitions.Any, new Message { Key = "ThisIsMyKey" }, new DateTime());
             var message2 = ProduceMessage.New(string.Empty, Partitions.Any, new Message { Key = "ThisIsMyOtherKey" }, new DateTime());
@@ -50,7 +51,7 @@ public void Test_MessageKeyPartitionSelection_Fallbacks_To_RoundRobin_If_Message
                 new Partition {Id = 1, Leader = nodeMock},
                 new Partition {Id = 2, Leader = nodeMock},
             };
-            var partitionStrategy = new MessageKeyPartitionSelection(Serializer, RoundRobinPartitionSelection);
+            var partitionStrategy = new MessageKeyPartitionSelection(Serializer, RoundRobinPartitionSelection, Mock.Of<ILogger>());
             var partitioner = new PartitionSelector(partitionStrategy);
             var message = ProduceMessage.New(string.Empty, Partitions.Any, new Message { Key = null }, new DateTime());
 
@@ -71,7 +72,7 @@ public void Test_MessageKeyPartitionSelection_Fallbacks_To_RoundRobin_If_Partiti
                 new Partition {Id = 2, Leader = nodeMock},
             };
             var blacklistedPartitions = new Dictionary<int, DateTime> { { partitionIdBlacklisted, DateTime.MaxValue } };
-            var partitionStrategy = new MessageKeyPartitionSelection(Serializer, RoundRobinPartitionSelection);
+            var partitionStrategy = new MessageKeyPartitionSelection(Serializer, RoundRobinPartitionSelection, Mock.Of<ILogger>());
             var partitioner = new PartitionSelector(partitionStrategy);
             var message = ProduceMessage.New(string.Empty, Partitions.Any, new Message { Key = "ThisIsMyKey" }, new DateTime());
 
diff --git a/kafka-sharp/kafka-sharp/Public/PartitionSelectionConfig.cs b/kafka-sharp/kafka-sharp/Public/PartitionSelectionConfig.cs
index eff5985..3ddc3dc 100644
--- a/kafka-sharp/kafka-sharp/Public/PartitionSelectionConfig.cs
+++ b/kafka-sharp/kafka-sharp/Public/PartitionSelectionConfig.cs
@@ -40,7 +40,8 @@ public void SetPartitionSelectionStrategyForTopic(string topic, PartitionSelecti
             _selectionStrategyByTopic.Add(topic, strategy);
         }
 
-        internal IPartitionSelection GetPartitionSelectionForTopic(string topic, int delay, int startSeed, ISerializer keySerializer)
+        internal IPartitionSelection GetPartitionSelectionForTopic(string topic, int delay, int startSeed,
+            ISerializer keySerializer, ILogger logger)
         {
             if (!_selectionStrategyByTopic.TryGetValue(topic, out var selectionStrategy))
             {
@@ -53,7 +54,7 @@ internal IPartitionSelection GetPartitionSelectionForTopic(string topic, int del
                 return roundRobinSelection;
             }
 
-            return new MessageKeyPartitionSelection(keySerializer, roundRobinSelection);
+            return new MessageKeyPartitionSelection(keySerializer, roundRobinSelection, logger);
         }
     }
 }
diff --git a/kafka-sharp/kafka-sharp/Routing/PartitionSelection/MessageKeyPartitionSelection.cs b/kafka-sharp/kafka-sharp/Routing/PartitionSelection/MessageKeyPartitionSelection.cs
index 01a5cce..c2b0bcc 100644
--- a/kafka-sharp/kafka-sharp/Routing/PartitionSelection/MessageKeyPartitionSelection.cs
+++ b/kafka-sharp/kafka-sharp/Routing/PartitionSelection/MessageKeyPartitionSelection.cs
@@ -9,18 +9,20 @@ namespace Kafka.Routing.PartitionSelection
 {
     internal class MessageKeyPartitionSelection : IPartitionSelection
     {
-        private static readonly ILogger Logger = LoggerManager.Instance;
         private const int LogSamplingPercentage = 99;
 
         private readonly ISerializer _keySerializer;
         private readonly IPartitionSelection _roundRobinSelection;
+        private readonly ILogger _logger;
         private readonly Pool<ReusableMemoryStream> _messageKeyBuffersPool;
         private readonly Random _rnd;
 
-        public MessageKeyPartitionSelection(ISerializer keySerializer, IPartitionSelection roundRobinSelection)
+        public MessageKeyPartitionSelection(ISerializer keySerializer, IPartitionSelection roundRobinSelection,
+            ILogger logger)
         {
             _keySerializer = keySerializer;
             _roundRobinSelection = roundRobinSelection;
+            _logger = logger;
             _rnd = new Random();
             _messageKeyBuffersPool = new Pool<ReusableMemoryStream>(
                 limit: 100,
@@ -61,7 +63,7 @@ private int GetPartitionIdFromKey(object messageKey, int partitionsLength)
 
             if (_rnd.Next(0, 100) >= LogSamplingPercentage)
             {
-                Logger.LogError($"{typeof(MessageKeyPartitionSelection)} cannot determine partition as message's "
+                _logger.LogError($"{typeof(MessageKeyPartitionSelection)} cannot determine partition as message's "
                     + "key is null. Falling back to round robin selection");
             }
 
diff --git a/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs b/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs
index 986be6b..48a96ad 100644
--- a/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs
+++ b/kafka-sharp/kafka-sharp/Routing/ProducerRouter.cs
@@ -568,7 +568,8 @@ private async Task HandleProduceMessage(ProduceMessage produceMessage)
             {
                 var keySerializer = _configuration.SerializationConfig.GetSerializersForTopic(topic).Item1;
                 var partitionSelectionImpl = _configuration.PartitionSelectionConfig.GetPartitionSelectionForTopic(
-                    topic, _configuration.NumberOfMessagesBeforeRoundRobin, _randomGenerator.Next(), keySerializer);
+                    topic, _configuration.NumberOfMessagesBeforeRoundRobin, _randomGenerator.Next(),
+                    keySerializer, _cluster.Logger);
                 selector = new PartitionSelector(partitionSelectionImpl);
                 _partitioners[topic] = selector;
             }