fix #160

RELEASE_NOTES.md

@@ -1,3 +1,7 @@
+### 0.1.4 - 03.08.2017
+
+* Fixed `FetchResponse` `MessageTooBigException` when a message set has been compressed. (#160)
+
 ### 0.1.4-beta - 25.07.2017
 
 * Fixed v0.10.1 protocol bug for `Offset` API.

src/kafunk/Compression.fs

@@ -31,7 +31,7 @@ module internal Stream =
  use compStream = makeStream inputStream
  compStream.CopyTo(outputStream)
  let buf = Binary.Segment(outputStream.GetBuffer(), 0, int outputStream.Length)
- MessageSet.Read (messageVer, 0, 0s, buf.Count, BinaryZipper(buf))
+ MessageSet.Read (messageVer, 0, 0s, buf.Count, true, BinaryZipper(buf))
 
 [<Compile(Module)>]
 module GZip =
@@ -151,7 +151,7 @@ module Snappy =
 
  let decompress (messageVer:ApiVersion) (m:Message) =
  let buf = CompressedMessage.decompress m.value 
- MessageSet.Read (messageVer, 0, 0s, buf.Count, BinaryZipper(buf))
+ MessageSet.Read (messageVer, 0, 0s, buf.Count, true, BinaryZipper(buf))
 
 let compress (messageVer:int16) (compression:byte) (ms:MessageSet) =
  match compression with

src/kafunk/Protocol.fs

@@ -434,7 +434,9 @@ module Protocol =
  buf.WriteInt32 x.messageSize
  Message.Write (messageVer, x.message, buf)
 
- static member internal Read (messageVer:ApiVersion, partition:Partition, ec:ErrorCode, messageSetSize:int, buf:BinaryZipper) =
+ // NB: skipTooLarge=true is for scenarios where decompression is involved and a message set is being decoded from an individual message
+ // which was itself too small.
+ static member internal Read (messageVer:ApiVersion, partition:Partition, ec:ErrorCode, messageSetSize:int, skipTooLarge:bool, buf:BinaryZipper) =
  let mutable consumed = 0
  let arr = ResizeArray<_>()
  while consumed < messageSetSize && buf.Buffer.Count > 0 do
@@ -445,7 +447,20 @@ module Protocol =
  let (messageSize:MessageSize) = buf.ReadInt32 ()
  let messageSetRemainder = messageSetRemainder - 12 // (Offset + MessageSize)
  if messageSize > messageSetSize then
- raise (MessageTooBigException(sprintf "partition=%i offset=%i message_set_size=%i message_size=%i" partition offset messageSetSize messageSize))
+ let errMsg = sprintf "partition=%i offset=%i message_set_size=%i message_size=%i consumed_bytes=%i consumed_count=%i" 
+ partition offset messageSetSize messageSize consumed arr.Count
+ if not skipTooLarge then
+ raise (MessageTooBigException(errMsg))
+ else
+// let payload = Binary.toString buf.Buffer
+// printfn "|WARN|MessageTooBig|%s" errMsg
+// printfn "|WARN|MessageTooBig|payload=%s" payload
+// try
+// let message = Message.Read (messageVer,buf)
+// printfn "|WARN|MessageTooBig|payload=%s" (Binary.toString message.value)
+// with ex ->
+// printfn "ERROR DECODING MESSAGE|%O" ex
+ ()
  try
  if messageSetRemainder >= messageSize && buf.Buffer.Count >= messageSize then
  let message = Message.Read (messageVer,buf)
@@ -740,7 +755,7 @@ module Protocol =
  let errorCode = buf.ReadInt16 ()
  let hwo = buf.ReadInt64 ()
  let mss = buf.ReadInt32 ()
- let ms = MessageSet.Read (MessageVersions.fetchResMessage ver,partition,errorCode,mss,buf)
+ let ms = MessageSet.Read (MessageVersions.fetchResMessage ver,partition,errorCode,mss,false,buf)
  ps.[j] <- partition, errorCode, hwo, mss, ms
  topics.[i] <- (t,ps)
  let res = FetchResponse(throttleTime, topics)

tests/kafunk.Tests/Consumer.fsx

@@ -5,7 +5,7 @@ open FSharp.Control
 open Kafunk
 open System
 
-Log.MinLevel <- LogLevel.Trace
+//Log.MinLevel <- LogLevel.Trace
 let Log = Log.create __SOURCE_FILE__
 
 let argiDefault i def = fsi.CommandLineArgs |> Seq.tryItem i |> Option.getOr def
@@ -20,18 +20,18 @@ let go = async {
  let connConfig = 
  let chanConfig = 
  ChanConfig.create (
- requestTimeout = TimeSpan.FromSeconds 30.0,
- receiveBufferSize = 8192 * 20,
- sendBufferSize = 8192 * 10,
+ requestTimeout = TimeSpan.FromSeconds 60.0,
+ receiveBufferSize = 8192 * 50,
+ sendBufferSize = 8192 * 50,
  connectRetryPolicy = ChanConfig.DefaultConnectRetryPolicy,
  requestRetryPolicy = ChanConfig.DefaultRequestRetryPolicy)
  KafkaConfig.create (
  [KafkaUri.parse host], 
  //[KafkaUri.parse "localhost:9092" ; KafkaUri.parse "localhost:9093" ; KafkaUri.parse "localhost:9094"],
  tcpConfig = chanConfig,
  requestRetryPolicy = KafkaConfig.DefaultRequestRetryPolicy,
- version = Versions.V_0_10_1,
- autoApiVersions = true)
+ version = Versions.V_0_9_0,
+ autoApiVersions = false)
  Kafka.connAsync connConfig
  let consumerConfig = 
  ConsumerConfig.create (