redo(ticdc): handle error correctly in redo manager

cdc/redo/manager.go

@@ -535,7 +535,6 @@ func (m *ManagerImpl) bgUpdateLog(ctx context.Context, errCh chan<- error) {
 				err := m.writer.WriteLog(ctx, cache.tableID, logs)
 				if err != nil {
 					handleErr(err)
-					return
 				}
 				m.metricWriteLogDuration.Observe(time.Since(start).Seconds())
 			case model.MessageTypeResolved:

cdc/redo/manager_test.go

@@ -23,6 +23,8 @@ import (
 
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/cdc/redo/writer"
+	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 )
@@ -310,3 +312,61 @@ func TestManagerRtsMap(t *testing.T) {
 	logMgr.postFlush(tables, minTs)
 	require.Equal(t, uint64(20), logMgr.GetMinResolvedTs())
 }
+
+// TestManagerError tests whether internal error in bgUpdateLog could be managed correctly.
+func TestManagerError(t *testing.T) {
+	t.Parallel()
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
+	defer cancel()
+
+	cfg := &config.ConsistentConfig{
+		Level:   string(ConsistentLevelEventual),
+		Storage: "blackhole://",
+	}
+	errCh := make(chan error, 1)
+	opts := &ManagerOptions{
+		EnableBgRunner: false,
+		ErrCh:          errCh,
+	}
+	logMgr, err := NewManager(ctx, cfg, opts)
+	require.Nil(t, err)
+	logMgr.writer = writer.NewInvalidBlackHoleWriter(logMgr.writer)
+	defer logMgr.Cleanup(ctx)
+
+	go logMgr.bgUpdateLog(ctx, errCh)
+
+	testCases := []struct {
+		tableID model.TableID
+		rows    []*model.RowChangedEvent
+	}{
+		{
+			tableID: 53,
+			rows: []*model.RowChangedEvent{
+				{CommitTs: 120, Table: &model.TableName{TableID: 53}},
+				{CommitTs: 125, Table: &model.TableName{TableID: 53}},
+				{CommitTs: 130, Table: &model.TableName{TableID: 53}},
+			},
+		},
+	}
+	for _, tc := range testCases {
+		err := logMgr.EmitRowChangedEvents(ctx, tc.tableID, tc.rows...)
+		require.Nil(t, err)
+	}
+
+	select {
+	case <-ctx.Done():
+		t.Fatal("bgUpdateLog should return error before context is done")
+	case err := <-errCh:
+		require.Regexp(t, ".*invalid black hole writer.*", err)
+		require.Regexp(t, ".*WriteLog.*", err)
+	}
+
+	go logMgr.bgUpdateLog(ctx, errCh)
+	select {
+	case <-ctx.Done():
+		t.Fatal("bgUpdateLog should return error before context is done")
+	case err := <-errCh:
+		require.Regexp(t, ".*invalid black hole writer.*", err)
+		require.Regexp(t, ".*FlushLog.*", err)
+	}
+}

cdc/redo/writer/blackhole_writer.go

@@ -15,6 +15,7 @@ package writer
 
 import (
 	"context"
+	"errors"
 	"sync"
 
 	"github.com/pingcap/log"
@@ -69,3 +70,26 @@ func (bs *blackHoleWriter) SendDDL(_ context.Context, ddl *model.RedoDDLEvent) e
 func (bs *blackHoleWriter) Close() error {
 	return nil
 }
+
+type invalidBlackHoleWriter struct {
+	*blackHoleWriter
+}
+
+// NewInvalidBlackHoleWriter creates a invalid blackHole writer
+func NewInvalidBlackHoleWriter(rl RedoLogWriter) *invalidBlackHoleWriter {
+	return &invalidBlackHoleWriter{
+		blackHoleWriter: rl.(*blackHoleWriter),
+	}
+}
+
+func (ibs *invalidBlackHoleWriter) WriteLog(
+	_ context.Context, _ model.TableID, _ []*model.RedoRowChangedEvent,
+) (err error) {
+	return errors.New("[WriteLog] invalid black hole writer")
+}
+
+func (ibs *invalidBlackHoleWriter) FlushLog(
+	_ context.Context, _, _ model.Ts,
+) error {
+	return errors.New("[FlushLog] invalid black hole writer")
+}