From a3aa3a071c80aa4ce5d89fd6cc20f40590af7e1f Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Wed, 1 Jan 2025 10:21:10 +0200
Subject: [PATCH] Fix EOF check

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 .../tabletmanager/vreplication/vplayer.go        |  4 ++--
 .../vreplication/vplayer_parallel_worker_test.go | 16 ++++++++++++++++
 2 files changed, 18 insertions(+), 2 deletions(-)

diff --git a/go/vt/vttablet/tabletmanager/vreplication/vplayer.go b/go/vt/vttablet/tabletmanager/vreplication/vplayer.go
index 81e765fe2b4..ab6cb9bec8f 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/vplayer.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/vplayer.go
@@ -290,7 +290,7 @@ func (vp *vplayer) fetchAndApply(ctx context.Context) (err error) {
 		// is shutting down and canceled the context, or stop position was reached,
 		// or a journal event was encountered.
 		// If so, we return nil which will cause the controller to not retry.
-		if errors.Is(err, io.EOF) {
+		if errors.Is(vterrors.UnwrapAll(err), io.EOF) {
 			return nil
 		}
 		return err
@@ -308,7 +308,7 @@ func (vp *vplayer) fetchAndApply(ctx context.Context) (err error) {
 		}
 		// If the stream ends normally we have to return an error indicating
 		// that the controller has to retry a different vttablet.
-		if err == nil || errors.Is(err, io.EOF) {
+		if err == nil || errors.Is(vterrors.UnwrapAll(err), io.EOF) {
 			return errors.New("vstream ended")
 		}
 		return err
diff --git a/go/vt/vttablet/tabletmanager/vreplication/vplayer_parallel_worker_test.go b/go/vt/vttablet/tabletmanager/vreplication/vplayer_parallel_worker_test.go
index 2def6f2ec62..66a684918f5 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/vplayer_parallel_worker_test.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/vplayer_parallel_worker_test.go
@@ -18,6 +18,7 @@ package vreplication
 
 import (
 	"errors"
+	"io"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -25,6 +26,21 @@ import (
 	"vitess.io/vitess/go/vt/vterrors"
 )
 
+func TestEOF(t *testing.T) {
+	{
+		err := io.EOF
+		assert.True(t, errors.Is(err, io.EOF))
+		unwrapped := vterrors.UnwrapAll(err)
+		assert.True(t, errors.Is(unwrapped, io.EOF))
+	}
+	{
+		err := vterrors.Wrapf(io.EOF, "unexpected EOF on table %s", "stress_test")
+		assert.False(t, errors.Is(err, io.EOF))
+		unwrapped := vterrors.UnwrapAll(err)
+		assert.True(t, errors.Is(unwrapped, io.EOF))
+	}
+}
+
 func TestErrRetryEvent(t *testing.T) {
 	err := vterrors.Wrapf(errRetryEvent, "unexpected event on table %s", "stress_test")
 	assert.True(t, errors.Is(vterrors.UnwrapAll(err), errRetryEvent))