fix/csv-scanner-infinite-loop

internal/impl/pure/scanner_csv.go

@@ -41,7 +41,7 @@ This scanner adds the following metadata to each message:
 				Description("If set to `true`, a quote may appear in an unquoted field and a non-doubled quote may appear in a quoted field.").
 				Default(false),
 			service.NewBoolField(scsvFieldContinueOnError).
-				Description("If a row fails to parse due to any error emit an empty message marked with the error and then continue consuming subsequent rows when possible. This can sometimes be useful in situations where input data contains individual rows which are malformed. However, when a row encounters a parsing error it is impossible to guarantee that following rows are valid, as this indicates that the input data is unreliable and could potentially emit misaligned rows.").
+				Description("If a row fails to parse emit an empty message marked with the error and then continue consuming subsequent rows when possible. Errors reading from the underlying source are not skipped and are returned as normal. This can sometimes be useful in situations where input data contains individual rows which are malformed. However, when a row encounters a parsing error it is impossible to guarantee that following rows are valid, as this indicates that the input data is unreliable and could potentially emit misaligned rows.").
 				Default(false),
 		)
 }
@@ -124,7 +124,11 @@ func (c *csvScanner) NextBatch(ctx context.Context) (service.MessageBatch, error
 
 	recordStrs, err := c.c.Read()
 	if err != nil {
-		if errors.Is(err, io.EOF) || !c.continueOnError {
+		// Only parse errors can be skipped. An error from the underlying reader
+		// (e.g. a broken network stream) repeats on every subsequent read, so
+		// continuing would emit empty messages forever.
+		var pErr *csv.ParseError
+		if !c.continueOnError || !errors.As(err, &pErr) {
 			return nil, err
 		}
 	}

internal/impl/pure/scanner_csv_test.go

@@ -3,7 +3,13 @@
 package pure_test
 
 import (
+	"context"
+	"encoding/csv"
+	"errors"
+	"io"
+	"strings"
 	"testing"
+	"testing/iotest"
 
 	"github.com/stretchr/testify/require"
 
@@ -83,3 +89,51 @@ a4,b4,c4
 		`["a4","b4","c4"]`,
 	)
 }
+
+func TestCSVScannerContinueOnError(t *testing.T) {
+	confSpec := service.NewConfigSpec().Field(service.NewScannerField("test"))
+	pConf, err := confSpec.ParseYAML(`
+test:
+  csv:
+    continue_on_error: true
+`, nil)
+	require.NoError(t, err)
+
+	creator, err := pConf.FieldScanner("test")
+	require.NoError(t, err)
+
+	readErr := errors.New("connection reset by peer")
+	rdr := io.MultiReader(
+		strings.NewReader("a,b\na1,b1\na2\na3,b3\n"),
+		iotest.ErrReader(readErr),
+	)
+
+	s, err := creator.Create(io.NopCloser(rdr), func(context.Context, error) error { return nil }, nil)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = s.Close(t.Context()) })
+
+	next := func() (service.MessageBatch, error) {
+		b, _, err := s.NextBatch(t.Context())
+		return b, err
+	}
+
+	b, err := next()
+	require.NoError(t, err)
+	require.Len(t, b, 1)
+	require.NoError(t, b[0].GetError())
+
+	// A parse error (wrong field count) is skipped as an errored message.
+	b, err = next()
+	require.NoError(t, err)
+	require.Len(t, b, 1)
+	require.ErrorIs(t, b[0].GetError(), csv.ErrFieldCount)
+
+	b, err = next()
+	require.NoError(t, err)
+	require.Len(t, b, 1)
+	require.NoError(t, b[0].GetError())
+
+	// A read error from the source is returned instead of looping forever.
+	_, err = next()
+	require.ErrorIs(t, err, readErr)
+}