Repository: redpanda-data/connect (language: go)

Issue:
Output codec without delimiter
Looking at the [available output codecs](https://www.benthos.dev/docs/components/outputs/file), I noticed that `all-bytes` [truncates](https://github.com/Jeffail/benthos/blob/d10ce3d96f0f313c79af7a7264f016b78d295c5f/internal/codec/writer.go#L71) on each message and `delim` [doesn't support](https://github.com/Jeffail/benthos/blob/d10ce3d96f0f313c79af7a7264f016b78d295c5f/internal/codec/writer.go#L58-L60) an empty delimiter.

Would you consider removing the non-empty delimiter restriction in the `delim` output codec? I need this when using the `chunker` input codec that I added in #640 for the use case of transferring a large file from Azure to a local file.


Discussion / hints:
Ahh of course, rather than allow empty delimiters it's probably better for discovery if we add an explicit `append` codec, that'll simply write the raw data.

Edit: but I'd also say it's fine to simply instantiate a `delim` codec with an empty delimiter under the hood to make the implementation simpler.
`append`. Of course! I was struggling to figure out what to name it 😁  Coming right up!

Fix patch:
diff --git a/internal/codec/reader.go b/internal/codec/reader.go
index 747780bb05..933dfcf8da 100644
--- a/internal/codec/reader.go
+++ b/internal/codec/reader.go
@@ -499,17 +499,23 @@ func (a *chunkerReader) Next(ctx context.Context) (types.Part, ReaderAckFn, erro
 	}
 
 	n, err := a.r.Read(a.buf)
-	if err == nil {
+	if err != nil {
+		if err == io.EOF {
+			a.finished = true
+		} else {
+			a.sourceAck(ctx, err)
+			return nil, nil, err
+		}
+	}
+
+	if n > 0 {
 		a.pending++
 
 		bytesCopy := make([]byte, n)
 		copy(bytesCopy, a.buf)
 		return message.NewPart(bytesCopy), a.ack, nil
-	} else if err == io.EOF {
-		a.finished = true
-	} else {
-		a.sourceAck(ctx, err)
 	}
+
 	return nil, nil, err
 }
 
diff --git a/internal/codec/writer.go b/internal/codec/writer.go
index 7556793736..a174e511b6 100644
--- a/internal/codec/writer.go
+++ b/internal/codec/writer.go
@@ -16,6 +16,7 @@ var WriterDocs = docs.FieldCommon(
 	"codec", "The way in which the bytes of messages should be written out into the output file. It's possible to write lines using a custom delimiter with the `delim:x` codec, where x is the character sequence custom delimiter.", "lines", "delim:\t", "delim:foobar",
 ).HasAnnotatedOptions(
 	"all-bytes", "Write the message to the file in full. If the file already exists the old content is deleted.",
+	"append", "Append messages to the file.",
 	"lines", "Append messages to the file followed by a line break.",
 	"delim:x", "Append messages to the file followed by a custom delimiter.",
 )
@@ -50,6 +51,10 @@ func GetWriter(codec string) (WriterConstructor, WriterConfig, error) {
 		return func(w io.WriteCloser) (Writer, error) {
 			return &allBytesWriter{w}, nil
 		}, allBytesConfig, nil
+	case "append":
+		return func(w io.WriteCloser) (Writer, error) {
+			return newCustomDelimWriter(w, "")
+		}, customDelimConfig, nil
 	case "lines":
 		return newLinesWriter, linesWriterConfig, nil
 	}


Test patch:
diff --git a/internal/codec/reader_test.go b/internal/codec/reader_test.go
index 8f3f4327b7..0a30952738 100644
--- a/internal/codec/reader_test.go
+++ b/internal/codec/reader_test.go
@@ -17,6 +17,20 @@ import (
 
 type noopCloser struct {
 	io.Reader
+	returnEOFOnRead bool
+}
+
+func (n noopCloser) Read(p []byte) (int, error) {
+	byteCount, err := n.Reader.Read(p)
+	if err != nil {
+		return byteCount, err
+	}
+
+	if n.returnEOFOnRead {
+		return byteCount, io.EOF
+	}
+
+	return byteCount, err
 }
 
 func (n noopCloser) Close() error {
@@ -25,7 +39,7 @@ func (n noopCloser) Close() error {
 
 func testReaderSuite(t *testing.T, codec, path string, data []byte, expected ...string) {
 	t.Run("close before reading", func(t *testing.T) {
-		buf := noopCloser{bytes.NewReader(data)}
+		buf := noopCloser{bytes.NewReader(data), false}
 
 		ctor, err := GetReader(codec, NewReaderConfig())
 		require.NoError(t, err)
@@ -42,8 +56,46 @@ func testReaderSuite(t *testing.T, codec, path string, data []byte, expected ...
 		assert.EqualError(t, ack, "service shutting down")
 	})
 
+	t.Run("returns all data even if EOF is encountered during the last read", func(t *testing.T) {
+		buf := noopCloser{bytes.NewReader(data), false}
+
+		ctor, err := GetReader(codec, NewReaderConfig())
+		require.NoError(t, err)
+
+		ack := errors.New("default err")
+
+		r, err := ctor(path, &buf, func(ctx context.Context, err error) error {
+			ack = err
+			return nil
+		})
+		require.NoError(t, err)
+
+		allReads := map[string][]byte{}
+
+		for i, exp := range expected {
+			if i == len(expected)-1 {
+				buf.returnEOFOnRead = true
+			}
+			p, ackFn, err := r.Next(context.Background())
+			require.NoError(t, err)
+			require.NoError(t, ackFn(context.Background(), nil))
+			assert.Equal(t, exp, string(p.Get()))
+			allReads[string(p.Get())] = p.Get()
+		}
+
+		_, _, err = r.Next(context.Background())
+		assert.EqualError(t, err, "EOF")
+
+		assert.NoError(t, r.Close(context.Background()))
+		assert.NoError(t, ack)
+
+		for k, v := range allReads {
+			assert.Equal(t, k, string(v), "Must not corrupt previous reads")
+		}
+	})
+
 	t.Run("acks ordered reads", func(t *testing.T) {
-		buf := noopCloser{bytes.NewReader(data)}
+		buf := noopCloser{bytes.NewReader(data), false}
 
 		ctor, err := GetReader(codec, NewReaderConfig())
 		require.NoError(t, err)
@@ -78,7 +130,7 @@ func testReaderSuite(t *testing.T, codec, path string, data []byte, expected ...
 	})
 
 	t.Run("acks unordered reads", func(t *testing.T) {
-		buf := noopCloser{bytes.NewReader(data)}
+		buf := noopCloser{bytes.NewReader(data), false}
 
 		ctor, err := GetReader(codec, NewReaderConfig())
 		require.NoError(t, err)
@@ -118,7 +170,7 @@ func testReaderSuite(t *testing.T, codec, path string, data []byte, expected ...
 	})
 
 	t.Run("acks parallel reads", func(t *testing.T) {
-		buf := noopCloser{bytes.NewReader(data)}
+		buf := noopCloser{bytes.NewReader(data), false}
 
 		ctor, err := GetReader(codec, NewReaderConfig())
 		require.NoError(t, err)
@@ -164,7 +216,7 @@ func testReaderSuite(t *testing.T, codec, path string, data []byte, expected ...
 
 	if len(expected) > 0 {
 		t.Run("nacks unordered reads", func(t *testing.T) {
-			buf := noopCloser{bytes.NewReader(data)}
+			buf := noopCloser{bytes.NewReader(data), false}
 
 			ctor, err := GetReader(codec, NewReaderConfig())
 			require.NoError(t, err)
@@ -275,6 +327,9 @@ func TestChunkerReader(t *testing.T) {
 	data := []byte("foobarbaz")
 	testReaderSuite(t, "chunker:3", "", data, "foo", "bar", "baz")
 
+	data = []byte("fooxbarybaz")
+	testReaderSuite(t, "chunker:3", "", data, "foo", "xba", "ryb", "az")
+
 	data = []byte("")
 	testReaderSuite(t, "chunker:1", "", data)
 }