Fix EOF check

Signed-off-by: Shlomi Noach <[email protected]>

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

go/vt/vttablet/tabletmanager/vreplication/vplayer_parallel_worker_test.go

@@ -18,13 +18,29 @@ package vreplication
 
 import (
 	"errors"
+	"io"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 
 	"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))