feat(kafka source): use msg.payload_len() to initialize FramedRead

changelog.d/kafka-source-parse_message.feature.md

@@ -0,0 +1,3 @@
+The `kafka` source will now consume less memory by not over allocating buffers while reading.
+
+authors: biuboombiuboom

src/sources/kafka.rs

@@ -991,7 +991,7 @@ fn parse_stream<'a>(
 
     let payload = Cursor::new(Bytes::copy_from_slice(payload));
 
-    let mut stream = FramedRead::new(payload, decoder);
+    let mut stream = FramedRead::with_capacity(payload, decoder, msg.payload_len());
     let (count, _) = stream.size_hint();
     let stream = stream! {
         while let Some(result) = stream.next().await {