small adjustments to fatal errors (#1830)

pkg/foundation/cerrors/fatal.go

@@ -24,7 +24,13 @@ type fatalError struct {
 }
 
 // FatalError creates a new fatalError.
-func FatalError(err error) *fatalError {
+func FatalError(err error) error {
+	if err == nil {
+		return nil
+	}
+	if IsFatalError(err) {
+		return err // already a fatal error
+	}
 	return &fatalError{Err: err}
 }
 
@@ -35,10 +41,7 @@ func (f *fatalError) Unwrap() error {
 
 // Error returns the error message.
 func (f *fatalError) Error() string {
-	if f.Err == nil {
-		return ""
-	}
-	return fmt.Sprintf("fatal error: %v", f.Err)
+	return fmt.Sprintf("fatal error: %s", f.Err.Error())
 }
 
 // IsFatalError checks if the error is a fatalError.

pkg/foundation/cerrors/fatal_test.go

@@ -22,11 +22,17 @@ import (
 	"github.com/matryer/is"
 )
 
-func TestNewFatalError(t *testing.T) {
+func TestFatalError(t *testing.T) {
 	is := is.New(t)
 
 	err := cerrors.New("test error")
+
+	// wrapping the error multiple times should not change the error message
 	fatalErr := cerrors.FatalError(err)
+	fatalErr = cerrors.FatalError(fatalErr)
+	fatalErr = cerrors.FatalError(fatalErr)
+	fatalErr = cerrors.FatalError(fatalErr)
+
 	wantErr := fmt.Sprintf("fatal error: %v", err)
 
 	is.Equal(fatalErr.Error(), wantErr)
@@ -56,6 +62,16 @@ func TestIsFatalError(t *testing.T) {
 			err:  err,
 			want: false,
 		},
+		{
+			name: "when it's nil",
+			err:  nil,
+			want: false,
+		},
+		{
+			name: "when underlying is nil",
+			err:  cerrors.FatalError(nil),
+			want: false,
+		},
 	}
 
 	for _, tc := range testCases {
@@ -74,13 +90,3 @@ func TestUnwrap(t *testing.T) {
 
 	is.Equal(cerrors.Unwrap(fatalErr), err)
 }
-
-func TestFatalError(t *testing.T) {
-	is := is.New(t)
-
-	err := cerrors.New("test error")
-	fatalErr := cerrors.FatalError(err)
-	wantErr := fmt.Sprintf("fatal error: %v", err)
-
-	is.Equal(fatalErr.Error(), wantErr)
-}

pkg/pipeline/stream/dlq.go

@@ -68,7 +68,7 @@ func (n *DLQHandlerNode) ID() string {
 
 // Run runs the DLQ handler node until all components depending on this node
 // call Done. Dependents can be added or removed while the node is running.
-func (n *DLQHandlerNode) Run(ctx context.Context) (err error) {
+func (n *DLQHandlerNode) Run(ctx context.Context) error {
 	defer n.state.Set(nodeStateStopped)
 	n.window = newDLQWindow(n.WindowSize, n.WindowNackThreshold)
 
@@ -78,7 +78,7 @@ func (n *DLQHandlerNode) Run(ctx context.Context) (err error) {
 	handlerCtx, n.handlerCtxCancel = context.WithCancel(context.Background())
 	defer n.handlerCtxCancel()
 
-	err = n.Handler.Open(handlerCtx)
+	err := n.Handler.Open(handlerCtx)
 	if err != nil {
 		return cerrors.Errorf("could not open DLQ handler: %w", err)
 	}
@@ -128,7 +128,7 @@ func (n *DLQHandlerNode) Ack(msg *Message) {
 	n.window.Ack()
 }
 
-func (n *DLQHandlerNode) Nack(msg *Message, nackMetadata NackMetadata) (err error) {
+func (n *DLQHandlerNode) Nack(msg *Message, nackMetadata NackMetadata) error {
 	state, err := n.state.Watch(msg.Ctx,
 		csync.WatchValues(nodeStateRunning, nodeStateStopped, dlqHandlerNodeStateBroken))
 	if err != nil {
@@ -146,10 +146,19 @@ func (n *DLQHandlerNode) Nack(msg *Message, nackMetadata NackMetadata) (err erro
 
 	ok := n.window.Nack()
 	if !ok {
-		return cerrors.FatalError(cerrors.Errorf(
-			"DLQ nack threshold exceeded (%d/%d), original error: %w",
-			n.WindowNackThreshold, n.WindowSize, nackMetadata.Reason,
-		))
+		if n.WindowNackThreshold > 0 {
+			// if the threshold is greater than 0 the DLQ is enabled and we
+			// need to respect the threshold by stopping the pipeline with a
+			// fatal error
+			return cerrors.FatalError(
+				cerrors.Errorf(
+					"DLQ nack threshold exceeded (%d/%d), original error: %w",
+					n.WindowNackThreshold, n.WindowSize, nackMetadata.Reason,
+				),
+			)
+		}
+		// DLQ is disabled, we don't need to wrap the error message
+		return nackMetadata.Reason
 	}
 
 	defer func() {

pkg/pipeline/stream/processor.go

@@ -90,9 +90,9 @@ func (n *ProcessorNode) Run(ctx context.Context) error {
 			// make sure that we ack as many records as possible
 			// (here we simply nack all of them, which is always only one)
 			if nackErr := msg.Nack(err, n.ID()); nackErr != nil {
-				return nackErr
+				return cerrors.FatalError(nackErr)
 			}
-			return err
+			return cerrors.FatalError(err)
 		}
 
 		switch v := recsOut[0].(type) {
@@ -114,6 +114,12 @@ func (n *ProcessorNode) Run(ctx context.Context) error {
 			if err != nil {
 				return cerrors.FatalError(cerrors.Errorf("error executing processor: %w", err))
 			}
+		default:
+			err := cerrors.Errorf("processor returned unknown record type: %T", v)
+			if nackErr := msg.Nack(err, n.ID()); nackErr != nil {
+				return cerrors.FatalError(nackErr)
+			}
+			return cerrors.FatalError(err)
 		}
 	}
 }