Merge pull request #41 from criteo/syncApril2018

More resilient consumer + various fixes

Author: ychuzevi

appveyor.yml

@@ -15,7 +15,7 @@ before_build:
 build_script:
 - dotnet build "kafka-sharp/kafka-sharp-netstd.sln" -c %CONFIGURATION%
 after_build:
-- dotnet pack /p:PackageVersion="%APPVEYOR_REPO_TAG_NAME%" "kafka-sharp/kafka-sharp/Kafka.netstandard.csproj" -c %CONFIGURATION% --no-build -o %APPVEYOR_BUILD_FOLDER%\artifacts
+- cmd: IF "%APPVEYOR_REPO_TAG%" == "true" (dotnet pack /p:PackageVersion="%APPVEYOR_REPO_TAG_NAME%" "kafka-sharp/kafka-sharp/Kafka.netstandard.csproj" -c %CONFIGURATION% --no-build -o %APPVEYOR_BUILD_FOLDER%\artifacts)
 test_script:
 - dotnet test "kafka-sharp/kafka-sharp.UTest/kafka.UTest.netstandard.csproj" -c %CONFIGURATION%
 cache:

kafka-sharp/kafka-sharp.UTest/Kafka.UTest.csproj

@@ -63,6 +63,9 @@
     <Reference Include="Microsoft.CSharp" />
     <Reference Include="System.Data" />
     <Reference Include="System.Net.Http" />
+    <Reference Include="System.ValueTuple">
+      <HintPath>..\packages\System.ValueTuple.4.3.0\lib\netstandard1.0\System.ValueTuple.dll</HintPath>
+    </Reference>
     <Reference Include="System.Xml" />
   </ItemGroup>
   <ItemGroup>

kafka-sharp/kafka-sharp.UTest/Mocks.cs

@@ -367,9 +367,12 @@ class EchoConnectionMock : SuccessConnectionMock
         private static int _count;
         private readonly int _responseDelayMs;
 
+        private static ConcurrentQueue<Timer> _timers = new ConcurrentQueue<Timer>();
+
         public static void Reset()
         {
             _count = 1;
+            _timers = new ConcurrentQueue<Timer>();
         }
 
         public EchoConnectionMock(bool forceErrors = false, int responseDelayMs = 0)
@@ -406,11 +409,12 @@ public override Task SendAsync(int correlationId, ReusableMemoryStream buffer, b
                 if (_responseDelayMs > 0)
                 {
                     var tcs = new TaskCompletionSource<bool>();
-                    new Timer(_ =>
+                    var timer = new Timer(_ =>
                     {
                         OnResponse(correlationId, response);
                         tcs.SetResult(true);
                     }, null, _responseDelayMs, -1);
+                    _timers.Enqueue(timer);
                     return tcs.Task;
                 }
 

kafka-sharp/kafka-sharp.UTest/TestConsumer.cs

@@ -182,10 +182,21 @@ Mocks InitCluster()
             return new Mocks { Cluster = cluster, Node = node, Group = group, HeartbeatCalled = heartbeatEvent};
         }
 
-        private void WaitForCallToHeartbeat(Mocks mock)
+        private void WaitOneSecondMaxForEvent(string name, AutoResetEvent ev)
         {
-            if (!mock.HeartbeatCalled.WaitOne(TimeSpan.FromSeconds(1)))
-                Assert.Fail("We waited 1 sec for a hearbeat to happen, but it never did.");
+            if (!ev.WaitOne(TimeSpan.FromSeconds(1)))
+            {
+                Assert.Fail("We waited 1 sec for " + name + " to happen, but it never did.");
+            }
+        }
+
+        private async Task HeartbeatFinishedProcessing(Mocks mock, ConsumeRouter router)
+        {
+            // First we wait to be sure that a heartbeat has started being processed
+            WaitOneSecondMaxForEvent("heatbeat", mock.HeartbeatCalled);
+            // Then we wait to be sure that the current message is finished processing
+            // (this message being the heartbeat or a following message)
+            await router.StopProcessingTask();
         }
 
         [Test]
@@ -505,7 +516,7 @@ public void TestConsumer_ConsumerGroupStartConsume()
             mocks.Node.Verify(n => n.Fetch(It.IsAny<FetchMessage>()), Times.Once); // 1 partition with specific offset
             mocks.Node.Verify(n => n.Offset(It.IsAny<OffsetMessage>()), Times.Once); // 1 partition with offset -1
 
-            WaitForCallToHeartbeat(mocks);
+            WaitOneSecondMaxForEvent("heatbeat", mocks.HeartbeatCalled);
 
             mocks.Group.Verify(g => g.Heartbeat());
 
@@ -540,27 +551,30 @@ public void TestConsumer_ConsumerGroupStartConsume()
             }});
 
             mocks.Node.Verify(n => n.Fetch(It.IsAny<FetchMessage>()), Times.Exactly(2)); // response should have triggered one more fetch
-            if (!commitEvent.WaitOne(TimeSpan.FromSeconds(1)))
-                Assert.Fail("We waited 1 sec for a commit from the group, it did not happen");
+            WaitOneSecondMaxForEvent("commit", commitEvent);
             mocks.Group.Verify(g => g.Commit(It.IsAny<IEnumerable<TopicData<OffsetCommitPartitionData>>>())); // should have auto commited
 
             consumer.Stop().Wait();
         }
 
         [Test]
-        public void TestConsumer_ConsumerGroupRestartConsume()
+        public async Task TestConsumer_ConsumerGroupRestartConsume()
         {
             var mocks = InitCluster();
             var consumer = new ConsumeRouter(mocks.Cluster.Object,
                 new Configuration { TaskScheduler = new CurrentThreadTaskScheduler(), ConsumeBatchSize = 1 }, 1);
 
+            var consumerStartEvent = new AutoResetEvent(false);
+            var consumerStopEvent = new AutoResetEvent(false);
+            consumer.ConsumerStopped += () => { consumerStopEvent.Set(); };
+
             consumer.StartConsumeSubscription(mocks.Group.Object, new[] { "the topic" });
 
             mocks.Group.Verify(g => g.Join(It.IsAny<IEnumerable<string>>()), Times.Once);
             mocks.Node.Verify(n => n.Fetch(It.IsAny<FetchMessage>()), Times.Once); // 1 partition with specific offset
             mocks.Node.Verify(n => n.Offset(It.IsAny<OffsetMessage>()), Times.Once); // 1 partition with offset -1
 
-            WaitForCallToHeartbeat(mocks);
+            WaitOneSecondMaxForEvent("heatbeat", mocks.HeartbeatCalled);
 
             mocks.Group.Verify(g => g.Heartbeat());
 
@@ -594,9 +608,11 @@ public void TestConsumer_ConsumerGroupRestartConsume()
                     }
             }});
 
-            mocks.Node.Verify(n => n.Fetch(It.IsAny<FetchMessage>()), Times.Exactly(2)); // response should have triggered one more fetch
 
             consumer.StopConsume("the topic", Partitions.All, Offsets.Now);
+            WaitOneSecondMaxForEvent("stop", consumerStopEvent);
+            mocks.Node.Verify(n => n.Fetch(It.IsAny<FetchMessage>()),
+                Times.Exactly(2)); // response should have triggered one more fetch
 
             consumer.Acknowledge(new CommonAcknowledgement<FetchResponse>
             {
@@ -628,8 +644,10 @@ public void TestConsumer_ConsumerGroupRestartConsume()
                     }
             }});
 
+            consumer.ConsumerStarted += () => { consumerStartEvent.Set(); };
             consumer.StartConsume("the topic", Partitions.All, Offsets.Now);
 
+            WaitOneSecondMaxForEvent("start", consumerStartEvent);
             mocks.Node.Verify(n => n.Fetch(It.IsAny<FetchMessage>()), Times.Exactly(3));
 
             consumer.Stop().Wait();
@@ -657,7 +675,7 @@ public void TestConsumer_ConsumerGroupCommit()
 
             consumer.StartConsumeSubscription(mocks.Group.Object, new[] { "the topic" });
 
-            WaitForCallToHeartbeat(mocks);
+            WaitOneSecondMaxForEvent("heatbeat", mocks.HeartbeatCalled);
 
             consumer.Acknowledge(new CommonAcknowledgement<FetchResponse>
             {
@@ -730,7 +748,7 @@ public async Task TestConsumer_ConsumerGroupCommitAsync()
         }
 
         [Test]
-        public void TestConsumer_ConsumerGroupHeartbeatErrors()
+        public async Task TestConsumer_ConsumerGroupHeartbeatErrors()
         {
             var mocks = InitCluster();
             mocks.Group.SetupGet(g => g.Configuration)
@@ -743,7 +761,7 @@ public void TestConsumer_ConsumerGroupHeartbeatErrors()
 
             consumer.StartConsumeSubscription(mocks.Group.Object, new[] { "the topic" });
 
-            WaitForCallToHeartbeat(mocks);
+            await HeartbeatFinishedProcessing(mocks, consumer);
 
             // At least 2 Join (one on start, one on next heartbeat)
             mocks.Group.Verify(g => g.Join(It.IsAny<IEnumerable<string>>()), Times.AtLeast(2));
@@ -761,7 +779,7 @@ public void TestConsumer_ConsumerGroupHeartbeatErrors()
 
             consumer.StartConsumeSubscription(mocks.Group.Object, new[] { "the topic" });
 
-            WaitForCallToHeartbeat(mocks);
+            await HeartbeatFinishedProcessing(mocks, consumer);
 
             mocks.Group.Verify(g => g.Join(It.IsAny<IEnumerable<string>>()), Times.AtLeast(2));
             // No Commit tried in case of ""hard" heartbeat errors
@@ -823,7 +841,9 @@ public async Task TestConsumer_RaisesPartitionsRevokedOnRebalance()
         {
             var mocks = InitCluster();
 
-            mocks.Group.Setup(g => g.Heartbeat()).ReturnsAsync(ErrorCode.RebalanceInProgress);
+            mocks.Group.Setup(g => g.Heartbeat())
+                .ReturnsAsync(ErrorCode.RebalanceInProgress)
+                .Callback(() => mocks.HeartbeatCalled.Set());
 
             mocks.Group.SetupGet(g => g.Configuration).Returns(
                 new ConsumerGroupConfiguration { AutoCommitEveryMs = -1, SessionTimeoutMs = 10 });
@@ -841,7 +861,8 @@ public async Task TestConsumer_RaisesPartitionsRevokedOnRebalance()
             consumer.PartitionsRevoked += () => partitionsRevokedEventIsCalled = true;
 
             consumer.StartConsumeSubscription(mocks.Group.Object, new[] { "the topic" });
-            Thread.Sleep(20);
+
+            await HeartbeatFinishedProcessing(mocks, consumer);
 
             Assert.That(partitionsRevokedEventIsCalled, Is.True);
 

kafka-sharp/kafka-sharp.UTest/TestConsumerGroup.cs

@@ -29,6 +29,7 @@
     <PackageReference Include="System.Reactive.Interfaces" Version="3.1.1" />
     <PackageReference Include="System.Reactive.Linq" Version="3.1.1" />
     <PackageReference Include="System.Threading.Thread" Version="4.3.0" />
+    <PackageReference Include="System.ValueTuple" Version="4.3.0" />
   </ItemGroup>
 
   <PropertyGroup>

kafka-sharp/kafka-sharp.UTest/kafka.UTest.netstandard.csproj

@@ -1454,8 +1454,10 @@ private void CleanUpConnection(IConnection connection, Exception exception = nul
         private static readonly long[] NoOffset = new long[0];
 
         // Build an empty response from a given Fetch request with error set to LocalError.
-        private static FetchResponse BuildEmptyFetchResponseFromOriginal(
-            IBatchByTopic<FetchMessage> originalRequest)
+        // If the partition and topic match the parameters it means that the deserialization of a message
+        // for said (topic, partition) failed. The error is set to DeserializationError in this case.
+        private static FetchResponse BuildEmptyFetchResponseFromOriginal(IBatchByTopic<FetchMessage> originalRequest,
+            string faultyTopic = null, int faultyPartition = -1)
         {
             return new FetchResponse
             {
@@ -1473,7 +1475,9 @@ private static FetchResponse BuildEmptyFetchResponseFromOriginal(
                                                 fm =>
                                                     new FetchPartitionResponse
                                                     {
-                                                        ErrorCode = ErrorCode.LocalError,
+                                                        ErrorCode = b.Key == faultyTopic && fm.Partition == faultyPartition
+                                                            ? ErrorCode.DeserializationError
+                                                            : ErrorCode.LocalError,
                                                         HighWatermarkOffset = -1,
                                                         Partition = fm.Partition,
                                                         Messages = ResponseMessageListPool.EmptyList
@@ -1553,6 +1557,11 @@ private void ProcessFetchResponse(int correlationId, ReusableMemoryStream respon
                 response.Response = _serialization.DeserializeResponse<FetchResponse>(correlationId, responseData,
                     _configuration.Compatibility == Compatibility.V0_8_2 ? Basics.ApiVersion.V0 : Basics.ApiVersion.V2);
             }
+            catch (ProtocolException pex)
+            {
+                OnDecodeError(pex);
+                response.Response = BuildEmptyFetchResponseFromOriginal(originalRequest, pex.Topic, pex.Partition);
+            }
             catch (Exception ex)
             {
                 OnDecodeError(ex);

kafka-sharp/kafka-sharp/Cluster/Node.cs

@@ -230,5 +230,10 @@ enum ErrorCode : short
 
         // Local error, not from brokers
         LocalError = -42,
+
+        /// <summary>
+        /// Deserialization on one message failed (bad Crc, magic number or compression)
+        /// </summary>
+        DeserializationError = -43,
     }
 }

kafka-sharp/kafka-sharp/Protocol/Errors.cs

@@ -81,7 +81,15 @@ public void Deserialize(ReusableMemoryStream stream, object extra, Basics.ApiVer
             Partition = BigEndianConverter.ReadInt32(stream);
             ErrorCode = (ErrorCode) BigEndianConverter.ReadInt16(stream);
             HighWatermarkOffset = BigEndianConverter.ReadInt64(stream);
-            Messages = DeserializeMessageSet(stream, extra as Deserializers);
+            try
+            {
+                Messages = DeserializeMessageSet(stream, extra as Deserializers);
+            }
+            catch (ProtocolException pEx)
+            {
+                pEx.Partition = Partition;
+                throw;
+            }
         }
 
         internal static List<ResponseMessage> DeserializeMessageSet(ReusableMemoryStream stream, Deserializers deserializers)

kafka-sharp/kafka-sharp/Protocol/FetchResponse.cs

@@ -7,18 +7,20 @@
 
 namespace Kafka.Protocol
 {
-    enum MessageVersion
+    internal enum MessageVersion
     {
         V0 = 0,
         V1 = 1
     }
 
-    struct Message
+    internal struct Message
     {
         public object Key;
         public object Value;
         public long TimeStamp;
-        public ReusableMemoryStream SerializedKeyValue;
+
+        // Visible for tests
+        internal ReusableMemoryStream SerializedKeyValue;
 
         private const int MinimumValidSizeForSerializedKeyValue = 2 * 4; // At least 4 bytes for key size and 4 bytes for value size
 
@@ -30,6 +32,13 @@ public void SerializeKeyValue(ReusableMemoryStream target, Tuple<ISerializer, IS
             Value = null;
         }
 
+        public void ReleaseSerializedKeyValue()
+        {
+            // Make sure that the buffer cannot be disposed twice (not good for buffer pooling)
+            SerializedKeyValue?.Dispose();
+            SerializedKeyValue = null;
+        }
+
         public void Serialize(ReusableMemoryStream stream, CompressionCodec compressionCodec,
             Tuple<ISerializer, ISerializer> serializers, MessageVersion msgVersion)
         {

kafka-sharp/kafka-sharp/Protocol/Message.cs

@@ -5,6 +5,10 @@ namespace Kafka.Protocol
 {
     class ProtocolException : Exception
     {
+        // Fields to identify the faulty message (useful in case the exception stops a batch)
+        internal string Topic;
+        internal int Partition;
+
         public ProtocolException(string message) : base(message)
         {
         }

kafka-sharp/kafka-sharp/Protocol/ProtocolException.cs

@@ -37,12 +37,20 @@ public void Deserialize(ReusableMemoryStream stream, object extra, Basics.ApiVer
                 var config = extra as SerializationConfig;
                 pdExtra = config.GetDeserializersForTopic(TopicName);
             }
-            for (int i = 0; i < count; ++i)
+            try
             {
-                array[i] = new TPartitionData();
-                array[i].Deserialize(stream, pdExtra, version);
+                for (int i = 0; i < count; ++i)
+                {
+                    array[i] = new TPartitionData();
+                    array[i].Deserialize(stream, pdExtra, version);
+                }
+                PartitionsData = array;
+            }
+            catch (ProtocolException pEx)
+            {
+                pEx.Topic = TopicName;
+                throw;
             }
-            PartitionsData = array;
         }
 
         #endregion

kafka-sharp/kafka-sharp/Protocol/TopicData.cs

@@ -55,7 +55,7 @@ public enum Compatibility
     }
 
     /// <summary>
-    /// In case of network errors
+    /// In case of network or protocol errors
     /// </summary>
     public enum ErrorStrategy
     {
@@ -65,7 +65,8 @@ public enum ErrorStrategy
         Discard,
 
         /// <summary>
-        /// Retry sending messsages (this may end up in duplicate messages)
+        /// Retry sending messsages (this may end up in duplicate messages) for Producer.
+        /// Retry fetching messages (this may end up in an infinite loop) for Consumer.
         /// </summary>
         Retry
     }
@@ -142,10 +143,15 @@ public class Configuration
         public TimeSpan TemporaryIgnorePartitionTime = TimeSpan.FromSeconds(42);
 
         /// <summary>
-        /// Strategy in case opf network errors.
+        /// Strategy in case opf network errors for Producer.
         /// </summary>
         public ErrorStrategy ErrorStrategy = ErrorStrategy.Discard;
 
+        /// <summary>
+        /// Strategy in case of deserialization Error for Consumer.
+        /// </summary>
+        public ErrorStrategy ConsumerErrorStrategy = ErrorStrategy.Discard;
+
         /// <summary>
         /// Time slice for batching messages. We wait  that much time at most before processing
         /// a batch of messages.