cherry pick pingcap#1309 to release-4.0

Signed-off-by: ti-srebot <[email protected]>

cdc/processor.go

@@ -293,10 +293,7 @@ func (p *processor) Run(ctx context.Context) {
 
 	go func() {
 		if err := wg.Wait(); err != nil {
-			select {
-			case p.errCh <- err:
-			default:
-			}
+			p.sendError(err)
 		}
 	}()
 }
@@ -1005,12 +1002,20 @@ func (p *processor) addTable(ctx context.Context, tableID int64, replicaInfo *mo
 		// start table puller
 		enableOldValue := p.changefeed.Config.EnableOldValue
 		span := regionspan.GetTableSpan(tableID, enableOldValue)
+<<<<<<< HEAD
 		kvStorage := util.KVStorageFromCtx(ctx)
+=======
+		kvStorage, err := util.KVStorageFromCtx(ctx)
+		if err != nil {
+			p.sendError(err)
+			return nil
+		}
+>>>>>>> 405c23a... processor: fix error chan buffer full will block the whole server (#1309)
 		plr := puller.NewPuller(ctx, p.pdCli, p.credential, kvStorage, replicaInfo.StartTs, []regionspan.Span{span}, p.limitter, enableOldValue)
 		go func() {
 			err := plr.Run(ctx)
 			if errors.Cause(err) != context.Canceled {
-				p.errCh <- err
+				p.sendError(err)
 			}
 		}()
 
@@ -1024,11 +1029,11 @@ func (p *processor) addTable(ctx context.Context, tableID int64, replicaInfo *mo
 				if os.IsNotExist(errors.Cause(err)) {
 					err = os.MkdirAll(p.changefeed.SortDir, 0755)
 					if err != nil {
-						p.errCh <- errors.Annotate(cerror.WrapError(cerror.ErrProcessorSortDir, err), "create dir")
+						p.sendError(errors.Annotate(cerror.WrapError(cerror.ErrProcessorSortDir, err), "create dir"))
 						return nil
 					}
 				} else {
-					p.errCh <- errors.Annotate(cerror.WrapError(cerror.ErrProcessorSortDir, err), "sort dir check")
+					p.sendError(errors.Annotate(cerror.WrapError(cerror.ErrProcessorSortDir, err), "sort dir check"))
 					return nil
 				}
 			}
@@ -1040,15 +1045,15 @@ func (p *processor) addTable(ctx context.Context, tableID int64, replicaInfo *mo
 				sorterImpl = psorter.NewUnifiedSorter(p.changefeed.SortDir, tableName, util.CaptureAddrFromCtx(ctx))
 			}
 		default:
-			p.errCh <- cerror.ErrUnknownSortEngine.GenWithStackByArgs(p.changefeed.Engine)
+			p.sendError(cerror.ErrUnknownSortEngine.GenWithStackByArgs(p.changefeed.Engine))
 			return nil
 		}
 		sorter := puller.NewRectifier(sorterImpl, p.changefeed.GetTargetTs())
 
 		go func() {
 			err := sorter.Run(ctx)
 			if errors.Cause(err) != context.Canceled {
-				p.errCh <- err
+				p.sendError(err)
 			}
 		}()
 
@@ -1084,8 +1089,21 @@ func (p *processor) addTable(ctx context.Context, tableID int64, replicaInfo *mo
 	}
 
 	atomic.StoreUint64(&p.localResolvedTs, p.position.ResolvedTs)
+<<<<<<< HEAD
 	table.sorter = startPuller(tableID, &table.resolvedTs)
 
+=======
+	tableSink = startPuller(tableID, &table.resolvedTs, &table.checkpointTs)
+	table.cancel = func() {
+		cancel()
+		if tableSink != nil {
+			tableSink.Close()
+		}
+		if mTableSink != nil {
+			mTableSink.Close()
+		}
+	}
+>>>>>>> 405c23a... processor: fix error chan buffer full will block the whole server (#1309)
 	syncTableNumGauge.WithLabelValues(p.changefeedID, p.captureInfo.AdvertiseAddr).Inc()
 }
 
@@ -1116,7 +1134,7 @@ func (p *processor) sorterConsume(
 			select {
 			case <-ctx.Done():
 				if errors.Cause(ctx.Err()) != context.Canceled {
-					p.errCh <- ctx.Err()
+					p.sendError(ctx.Err())
 				}
 				return
 			case p.opDoneCh <- tableID:
@@ -1136,7 +1154,7 @@ func (p *processor) sorterConsume(
 		select {
 		case <-ctx.Done():
 			if errors.Cause(ctx.Err()) != context.Canceled {
-				p.errCh <- ctx.Err()
+				p.sendError(ctx.Err())
 			}
 			return
 		case pEvent := <-sorter.Output():
@@ -1148,7 +1166,7 @@ func (p *processor) sorterConsume(
 			select {
 			case <-ctx.Done():
 				if errors.Cause(ctx.Err()) != context.Canceled {
-					p.errCh <- ctx.Err()
+					p.sendError(ctx.Err())
 				}
 				return
 			case p.mounter.Input() <- pEvent:
@@ -1175,10 +1193,43 @@ func (p *processor) sorterConsume(
 					zap.Any("replicaInfo", replicaInfo),
 					zap.Any("row", pEvent))
 			}
+<<<<<<< HEAD
 			select {
 			case <-ctx.Done():
 				if errors.Cause(ctx.Err()) != context.Canceled {
 					p.errCh <- ctx.Err()
+=======
+			failpoint.Inject("ProcessorSyncResolvedError", func() {
+				p.errCh <- errors.New("processor sync resolved injected error")
+				failpoint.Return()
+			})
+			err := processRowChangedEvent(pEvent)
+			if err != nil {
+				if errors.Cause(err) != context.Canceled {
+					p.sendError(ctx.Err())
+				}
+				return
+			}
+		case <-globalResolvedTsReceiver.C:
+			localResolvedTs := atomic.LoadUint64(&p.localResolvedTs)
+			globalResolvedTs := atomic.LoadUint64(&p.globalResolvedTs)
+			var minTs uint64
+			if localResolvedTs < globalResolvedTs {
+				minTs = localResolvedTs
+				log.Warn("the local resolved ts is less than the global resolved ts",
+					zap.Uint64("localResolvedTs", localResolvedTs), zap.Uint64("globalResolvedTs", globalResolvedTs))
+			} else {
+				minTs = globalResolvedTs
+			}
+			if minTs == 0 || atomic.LoadUint64(&p.checkpointTs) == minTs {
+				continue
+			}
+
+			checkpointTs, err := sink.FlushRowChangedEvents(ctx, minTs)
+			if err != nil {
+				if errors.Cause(err) != context.Canceled {
+					p.errCh <- errors.Trace(err)
+>>>>>>> 405c23a... processor: fix error chan buffer full will block the whole server (#1309)
 				}
 				return
 			case p.output <- pEvent:
@@ -1202,7 +1253,7 @@ func (p *processor) pullerConsume(
 		select {
 		case <-ctx.Done():
 			if errors.Cause(ctx.Err()) != context.Canceled {
-				p.errCh <- ctx.Err()
+				p.sendError(ctx.Err())
 			}
 			return
 		case rawKV := <-plr.Output():
@@ -1268,8 +1319,15 @@ func runProcessor(
 		return nil, errors.Trace(err)
 	}
 	ctx, cancel := context.WithCancel(ctx)
+<<<<<<< HEAD
 	errCh := make(chan error, 1)
 	sink, err := sink.NewSink(ctx, changefeedID, info.SinkURI, filter, info.Config, opts, errCh)
+=======
+	// processor only receives one error from the channel, all producers to this
+	// channel must use the non-blocking way to send error.
+	errCh := make(chan error, 1)
+	s, err := sink.NewSink(ctx, changefeedID, info.SinkURI, filter, info.Config, opts, errCh)
+>>>>>>> 405c23a... processor: fix error chan buffer full will block the whole server (#1309)
 	if err != nil {
 		cancel()
 		return nil, errors.Trace(err)
@@ -1322,3 +1380,11 @@ func runProcessor(
 
 	return processor, nil
 }
+
+func (p *processor) sendError(err error) {
+	select {
+	case p.errCh <- err:
+	default:
+		log.Error("processor receives redundant error", zap.Error(err))
+	}
+}

cdc/sink/cdclog/file.go

@@ -376,6 +376,8 @@ func NewLocalFileSink(ctx context.Context, sinkURI *url.URL, errCh chan error) (
 			case <-ctx.Done():
 				return
 			case errCh <- err:
+			default:
+				log.Error("error channel is full", zap.Error(err))
 			}
 		}
 	}()

cdc/sink/cdclog/s3.go

@@ -381,6 +381,8 @@ func NewS3Sink(ctx context.Context, sinkURI *url.URL, errCh chan error) (*s3Sink
 			case <-ctx.Done():
 				return
 			case errCh <- err:
+			default:
+				log.Error("error channel is full", zap.Error(err))
 			}
 		}
 	}()

cdc/sink/mq.go

@@ -152,6 +152,8 @@ func newMqSink(
 			case <-ctx.Done():
 				return
 			case errCh <- err:
+			default:
+				log.Error("error channel is full", zap.Error(err))
 			}
 		}
 	}()

cdc/sink/producer/kafka/kafka.go

@@ -366,6 +366,8 @@ func NewKafkaSaramaProducer(ctx context.Context, address string, topic string, c
 			case <-ctx.Done():
 				return
 			case errCh <- err:
+			default:
+				log.Error("error channel is full", zap.Error(err))
 			}
 		}
 	}()

cdc/sink/producer/pulsar/producer.go

@@ -20,7 +20,9 @@ import (
 
 	"github.com/apache/pulsar-client-go/pulsar"
 	"github.com/pingcap/failpoint"
+	"github.com/pingcap/log"
 	cerror "github.com/pingcap/ticdc/pkg/errors"
+	"go.uber.org/zap"
 )
 
 // NewProducer create a pulsar producer.
@@ -83,6 +85,7 @@ func (p *Producer) errors(_ pulsar.MessageID, _ *pulsar.ProducerMessage, err err
 		select {
 		case p.errCh <- cerror.WrapError(cerror.ErrPulsarSendMessage, err):
 		default:
+			log.Error("error channel is full", zap.Error(err))
 		}
 	}
 }

tests/processor_err_chan/conf/diff_config.toml

@@ -0,0 +1,27 @@
+# diff Configuration.
+
+log-level = "info"
+chunk-size = 10
+check-thread-count = 4
+sample-percent = 100
+use-rowid = false
+use-checksum = true
+fix-sql-file = "fix.sql"
+
+# tables need to check.
+[[check-tables]]
+    schema = "processor_err_chan"
+    tables = ["~.*"]
+
+[[source-db]]
+    host = "127.0.0.1"
+    port = 4000
+    user = "root"
+    password = ""
+    instance-id = "source-1"
+
+[target-db]
+    host = "127.0.0.1"
+    port = 3306
+    user = "root"
+    password = ""

tests/processor_err_chan/run.sh

@@ -0,0 +1,70 @@
+#!/bin/bash
+
+set -e
+
+CUR=$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )
+source $CUR/../_utils/test_prepare
+WORK_DIR=$OUT_DIR/$TEST_NAME
+CDC_BINARY=cdc.test
+SINK_TYPE=$1
+
+function check_changefeed_mark_stopped() {
+    endpoints=$1
+    changefeedid=$2
+    error_msg=$3
+    info=$(cdc cli changefeed query --pd=$endpoints -c $changefeedid -s)
+    echo "$info"
+    state=$(echo $info|jq -r '.state')
+    if [[ ! "$state" == "stopped" ]]; then
+        echo "changefeed state $state does not equal to stopped"
+        exit 1
+    fi
+    message=$(echo $info|jq -r '.error.message')
+    if [[ ! "$message" =~ "$error_msg" ]]; then
+        echo "error message '$message' is not as expected '$error_msg'"
+        exit 1
+    fi
+}
+
+export -f check_changefeed_mark_stopped
+
+function run() {
+    rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+    start_tidb_cluster --workdir $WORK_DIR
+    cd $WORK_DIR
+
+    pd_addr="http://$UP_PD_HOST_1:$UP_PD_PORT_1"
+    TOPIC_NAME="ticdc-processor-err-chan-$RANDOM"
+    case $SINK_TYPE in
+        kafka) SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?partition-num=4";;
+        *) SINK_URI="mysql://[email protected]:3306/?max-txn-row=1";;
+    esac
+    if [ "$SINK_TYPE" == "kafka" ]; then
+      run_kafka_consumer $WORK_DIR "kafka://127.0.0.1:9092/$TOPIC_NAME?partition-num=4"
+    fi
+
+    run_sql "CREATE DATABASE processor_err_chan;" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+    run_sql "CREATE DATABASE processor_err_chan;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+    for i in $(seq 1 10); do
+        run_sql "CREATE table processor_err_chan.t$i (id int primary key auto_increment)" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+        run_sql "CREATE table processor_err_chan.t$i (id int primary key auto_increment)" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+    done
+
+    run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY --addr "127.0.0.1:8300" --pd $pd_addr
+    changefeed_id=$(cdc cli changefeed create --pd=$pd_addr --sink-uri="$SINK_URI" --sort-engine=abc-engine 2>&1|tail -n2|head -n1|awk '{print $2}')
+
+    retry_time=10
+    ensure $retry_time check_changefeed_mark_stopped $pd_addr $changefeed_id "[CDC:ErrUnknownSortEngine]unknown sort engine abc-engine"
+
+    cdc cli changefeed create --pd=$pd_addr --sink-uri="$SINK_URI"
+    for i in $(seq 1 10); do
+        run_sql "INSERT INTO processor_err_chan.t$i values (),(),(),(),(),(),()" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+    done
+    check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
+
+    cleanup_process $CDC_BINARY
+}
+
+trap stop_tidb_cluster EXIT
+run $*
+echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"