fix(codecs): Ensure that batches using newline delimited framing end in a newline

changelog.d/batch-newline.fix.md

@@ -0,0 +1,3 @@
+Batches encoded using newline delimited framing now end with a trailing newline.
+
+authors: jszwedko

src/codecs/encoding/encoder.rs

@@ -93,12 +93,14 @@ impl Encoder<Framer> {
     }
 
     /// Get the suffix that encloses a batch of events.
-    pub const fn batch_suffix(&self) -> &[u8] {
-        match (&self.framer, &self.serializer) {
+    pub const fn batch_suffix(&self, empty: bool) -> &[u8] {
+        match (&self.framer, &self.serializer, empty) {
             (
                 Framer::CharacterDelimited(CharacterDelimitedEncoder { delimiter: b',' }),
                 Serializer::Json(_) | Serializer::NativeJson(_),
+                _,
             ) => b"]",
+            (Framer::NewlineDelimited(_), _, false) => b"\n",
             _ => &[],
         }
     }
@@ -323,4 +325,23 @@ mod tests {
         let sink = framed.into_inner();
         assert_eq!(sink, b"(foo)(bar)");
     }
+
+    #[tokio::test]
+    async fn test_encode_batch_newline() {
+        let encoder = Encoder::<Framer>::new(
+            Framer::NewlineDelimited(NewlineDelimitedEncoder::default()),
+            TextSerializerConfig::default().build().into(),
+        );
+        let source = futures::stream::iter(vec![
+            Event::Log(LogEvent::from("bar")),
+            Event::Log(LogEvent::from("baz")),
+            Event::Log(LogEvent::from("bat")),
+        ])
+        .map(Ok);
+        let sink: Vec<u8> = Vec::new();
+        let mut framed = FramedWrite::new(sink, encoder);
+        source.forward(&mut framed).await.unwrap();
+        let sink = framed.into_inner();
+        assert_eq!(sink, b"bar\nbaz\nbat\n");
+    }
 }

src/sinks/util/encoding.rs

@@ -31,6 +31,7 @@ impl Encoder<Vec<Event>> for (Transformer, crate::codecs::Encoder<Framer>) {
         let mut encoder = self.1.clone();
         let mut bytes_written = 0;
         let mut n_events_pending = events.len();
+        let is_empty = events.is_empty();
         let batch_prefix = encoder.batch_prefix();
         write_all(writer, n_events_pending, batch_prefix)?;
         bytes_written += batch_prefix.len();
@@ -65,7 +66,7 @@ impl Encoder<Vec<Event>> for (Transformer, crate::codecs::Encoder<Framer>) {
             n_events_pending -= 1;
         }
 
-        let batch_suffix = encoder.batch_suffix();
+        let batch_suffix = encoder.batch_suffix(is_empty);
         assert!(n_events_pending == 0);
         write_all(writer, 0, batch_suffix)?;
         bytes_written += batch_suffix.len();
@@ -295,9 +296,9 @@ mod tests {
             .sum::<JsonSize>();
 
         let (written, json_size) = encoding.encode_input(input, &mut writer).unwrap();
-        assert_eq!(written, 15);
+        assert_eq!(written, 16);
 
-        assert_eq!(String::from_utf8(writer).unwrap(), r#"{"key":"value"}"#);
+        assert_eq!(String::from_utf8(writer).unwrap(), "{\"key\":\"value\"}\n");
         assert_eq!(CountByteSize(1, input_json_size), json_size.size().unwrap());
     }
 
@@ -332,11 +333,11 @@ mod tests {
             .sum::<JsonSize>();
 
         let (written, json_size) = encoding.encode_input(input, &mut writer).unwrap();
-        assert_eq!(written, 50);
+        assert_eq!(written, 51);
 
         assert_eq!(
             String::from_utf8(writer).unwrap(),
-            "{\"key\":\"value1\"}\n{\"key\":\"value2\"}\n{\"key\":\"value3\"}"
+            "{\"key\":\"value1\"}\n{\"key\":\"value2\"}\n{\"key\":\"value3\"}\n"
         );
         assert_eq!(CountByteSize(3, input_json_size), json_size.size().unwrap());
     }