Decode error when decoding records which sent from native kafka client

[iamazy]

In kafkas, when I consume records which produced from native kafka client, it always got Decode Error. but everything is OK when I use Producer in kafkas.

Currently I ignore the DecodeError with a little changes like this.

    pub fn decode<B: ByteBuf>(buf: &mut B) -> Result<Vec<Record>, DecodeError> {
        let mut records = Vec::new();
        while buf.has_remaining() {
-            Self::decode_batch(buf, &mut records)?;
+            if let Err(_) = Self::decode_batch(buf, &mut records) {
+               break;
+            }
        }
        Ok(records)
    }

There seems has no record lost.