diff --git a/flow/client/client.go b/flow/client/client.go
index 9e9403685..d74e4fd47 100644
--- a/flow/client/client.go
+++ b/flow/client/client.go
@@ -108,7 +108,7 @@ func (c *GRPCClient) Send(event *proto.FlowEvent) error {
 func (c *GRPCClient) Receive(ctx context.Context, interval time.Duration, msgHandler func(msg *proto.FlowEventAck) error) error {
 	backOff := defaultBackoff(ctx, interval)
 	operation := func() error {
-		if err := c.establishStreamAndReceive(ctx, msgHandler); err != nil {
+		if err := c.establishStreamAndReceive(ctx, msgHandler); err == nil {
 			if errors.Is(err, context.Canceled) {
 				return backoff.Permanent(err)
 			}
@@ -124,7 +124,7 @@ func (c *GRPCClient) Receive(ctx context.Context, interval time.Duration, msgHan
 						return err
 					}
 					log.Infof("connection recreated successfully")
-					return nil
+					return fmt.Errorf("connection recreated, re-establishing stream")
 				}
 			}