Merge branch 'release-5.0' into cherry-pick-3013-to-release-5.0

Makefile

@@ -135,10 +135,10 @@ integration_test_build: check_failpoint_ctl
 integration_test: integration_test_mysql
 
 integration_test_mysql:
-	tests/run.sh mysql "$(CASE)"
+	tests/integration_tests/run.sh mysql "$(CASE)"
 
 integration_test_kafka: check_third_party_binary
-	tests/run.sh kafka "$(CASE)"
+	tests/integration_tests/run.sh kafka "$(CASE)"
 
 fmt: tools/bin/gofumports tools/bin/shfmt
 	@echo "gofmt (simplify)"
@@ -176,16 +176,21 @@ tidy:
 
 check: check-copyright fmt lint check-static tidy errdoc check-leaktest-added check-merge-conflicts
 
-coverage:
+integration_test_coverage:
 	GO111MODULE=off go get github.com/wadey/gocovmerge
 	gocovmerge "$(TEST_DIR)"/cov.* | grep -vE ".*.pb.go|$(CDC_PKG)/testing_utils/.*|$(CDC_PKG)/cdc/kv/testing.go|$(CDC_PKG)/cdc/entry/schema_test_helper.go|$(CDC_PKG)/cdc/sink/simple_mysql_tester.go|.*.__failpoint_binding__.go" > "$(TEST_DIR)/all_cov.out"
-	grep -vE ".*.pb.go|$(CDC_PKG)/testing_utils/.*|$(CDC_PKG)/cdc/kv/testing.go|$(CDC_PKG)/cdc/sink/simple_mysql_tester.go|.*.__failpoint_binding__.go" "$(TEST_DIR)/cov.unit.out" > "$(TEST_DIR)/unit_cov.out"
 ifeq ("$(JenkinsCI)", "1")
 	GO111MODULE=off go get github.com/mattn/goveralls
 	@goveralls -coverprofile=$(TEST_DIR)/all_cov.out -service=jenkins-ci -repotoken $(COVERALLS_TOKEN)
-	@bash <(curl -s https://codecov.io/bash) -f $(TEST_DIR)/unit_cov.out -t $(CODECOV_TOKEN)
 else
 	go tool cover -html "$(TEST_DIR)/all_cov.out" -o "$(TEST_DIR)/all_cov.html"
+endif
+
+unit_test_coverage:
+	grep -vE ".*.pb.go|$(CDC_PKG)/testing_utils/.*|$(CDC_PKG)/cdc/kv/testing.go|$(CDC_PKG)/cdc/sink/simple_mysql_tester.go|.*.__failpoint_binding__.go" "$(TEST_DIR)/cov.unit.out" > "$(TEST_DIR)/unit_cov.out"
+ifeq ("$(JenkinsCI)", "1")
+	@bash <(curl -s https://codecov.io/bash) -f $(TEST_DIR)/unit_cov.out -t $(CODECOV_TOKEN)
+else
 	go tool cover -html "$(TEST_DIR)/unit_cov.out" -o "$(TEST_DIR)/unit_cov.html"
 	go tool cover -func="$(TEST_DIR)/unit_cov.out"
 endif

cdc/sink/producer/kafka/kafka.go

@@ -537,10 +537,21 @@ func newSaramaConfig(ctx context.Context, c *Config) (*sarama.Config, error) {
 	}
 	config.Version = version
 	// See: https://kafka.apache.org/documentation/#replication
-	// When one of the brokers in a Kafka cluster is down, the partition leaders in this broker is broken, Kafka will election a new partition leader and replication logs, this process will last from a few seconds to a few minutes. Kafka cluster will not provide a writing service in this process.
-	// Time out in one minute(120 * 500ms).
+	// When one of the brokers in a Kafka cluster is down, the partition leaders
+	// in this broker is broken, Kafka will election a new partition leader and
+	// replication logs, this process will last from a few seconds to a few minutes.
+	// Kafka cluster will not provide a writing service in this process.
+	// Time out in one minute.
 	config.Metadata.Retry.Max = 120
 	config.Metadata.Retry.Backoff = 500 * time.Millisecond
+	// If it is not set, this means a metadata request against an unreachable
+	// cluster (all brokers are unreachable or unresponsive) can take up to
+	// `Net.[Dial|Read]Timeout * BrokerCount * (Metadata.Retry.Max + 1) +
+	// Metadata.Retry.Backoff * Metadata.Retry.Max`
+	// to fail.
+	// See: https://github.com/Shopify/sarama/issues/765
+	// and https://github.com/pingcap/ticdc/issues/3352.
+	config.Metadata.Timeout = 1 * time.Minute
 
 	config.Producer.Partitioner = sarama.NewManualPartitioner
 	config.Producer.MaxMessageBytes = c.MaxMessageBytes
@@ -580,7 +591,6 @@ func newSaramaConfig(ctx context.Context, c *Config) (*sarama.Config, error) {
 			return nil, errors.Trace(err)
 		}
 	}
-
 	if c.SaslScram != nil && len(c.SaslScram.SaslUser) != 0 {
 		config.Net.SASL.Enable = true
 		config.Net.SASL.User = c.SaslScram.SaslUser

metrics/grafana/ticdc.json

@@ -1585,7 +1585,7 @@
               "refId": "A"
             },
             {
-              "expr": "sum(ticdc_processor_resolved_ts_lag{tidb_cluster=\"$tidb_cluster\", changefeed=~\"$changefeed\", capture=~\"$capture\"}) by (capture,changefeed)",
+              "expr": "sum(ticdc_processor_checkpoint_ts_lag{tidb_cluster=\"$tidb_cluster\", changefeed=~\"$changefeed\", capture=~\"$capture\"}) by (capture,changefeed)",
               "format": "time_series",
               "interval": "",
               "intervalFactor": 1,

tests/README.md → tests/integration_tests/README.md

@@ -36,8 +36,8 @@
 
 2. Run `make integration_test` to execute the integration tests. This command will
 
-    1. Check that all required executables exist.
-    2. Execute `tests/run.sh`
+   1. Check that all required executables exist.
+   2. Execute `tests/integration_tests/run.sh`
 
     > If want to run one integration test case only, just pass the CASE parameter, such as `make integration_test CASE=simple`.
 
@@ -50,4 +50,5 @@
 
 ## Writing new tests
 
-New integration tests can be written as shell scripts in `tests/TEST_NAME/run.sh`. The script should exit with a nonzero error code on failure.
+New integration tests can be written as shell scripts in `tests/integration_tests/TEST_NAME/run.sh`. The script should
+exit with a nonzero error code on failure.

tests/_utils/start_tidb_cluster_impl → tests/integration_tests/_utils/start_tidb_cluster_impl

@@ -10,6 +10,18 @@ OUT_DIR=
 tidb_config=
 pd_config=
 multiple_upstream_pd=
+random_file_name=
+
+# Random generate the sockets config.
+# Make sure we dont use the same sock.
+randomGenSocketsConf() {
+	date=$(date '+%s%N')
+
+	random_file_name="$OUT_DIR/tidb-config-$date.toml"
+
+	cat "$OUT_DIR/tidb-config.toml" >"$random_file_name"
+	echo "socket = \"/tmp/tidb-$date.sock\"" >>"$random_file_name"
+}
 
 while [[ ${1} ]]; do
 	case "${1}" in
@@ -196,26 +208,29 @@ tikv-server \
 sleep 2
 
 echo "Starting Upstream TiDB..."
+randomGenSocketsConf
 tidb-server \
 	-P ${UP_TIDB_PORT} \
-	-config "$OUT_DIR/tidb-config.toml" \
+	-config "$random_file_name" \
 	--store tikv \
 	--path ${UP_PD_HOST_1}:${UP_PD_PORT_1} \
 	--status=${UP_TIDB_STATUS} \
 	--log-file "$OUT_DIR/tidb.log" &
 
+randomGenSocketsConf
 tidb-server \
 	-P ${UP_TIDB_OTHER_PORT} \
-	-config "$OUT_DIR/tidb-config.toml" \
+	-config "$random_file_name" \
 	--store tikv \
 	--path ${UP_PD_HOST_1}:${UP_PD_PORT_1} \
 	--status=${UP_TIDB_OTHER_STATUS} \
 	--log-file "$OUT_DIR/tidb_other.log" &
 
 echo "Starting Downstream TiDB..."
+randomGenSocketsConf
 tidb-server \
 	-P ${DOWN_TIDB_PORT} \
-	-config "$OUT_DIR/tidb-config.toml" \
+	-config "$random_file_name" \
 	--store tikv \
 	--path ${DOWN_PD_HOST}:${DOWN_PD_PORT} \
 	--status=${DOWN_TIDB_STATUS} \
@@ -357,7 +372,7 @@ while ! curl -o /dev/null -sf http://127.0.0.1:17000/metrics 1>/dev/null 2>&1; d
 done
 
 echo "Starting CDC state checker..."
-cd $CUR/../../testing_utils/cdc_state_checker
+cd $CUR/../../../testing_utils/cdc_state_checker
 if [ ! -f ./cdc_state_checker ]; then
 	GO111MODULE=on go build
 fi

tests/_utils/start_tls_tidb_cluster_impl → tests/integration_tests/_utils/start_tls_tidb_cluster_impl

@@ -88,6 +88,7 @@ EOF
 
 # tidb server config file
 cat - >"$OUT_DIR/tidb-config-tls.toml" <<EOF
+socket = "/tmp/tidb-tls.sock"
 split-table = true
 alter-primary-key = true
 new_collations_enabled_on_first_bootstrap = true
@@ -141,7 +142,7 @@ run_sql "update mysql.tidb set variable_value='60m' where variable_name='tikv_gc
 	--ssl-key=$TLS_DIR/server-key.pem
 
 echo "Starting CDC state checker..."
-cd $CUR/../../testing_utils/cdc_state_checker
+cd $CUR/../../../testing_utils/cdc_state_checker
 if [ ! -f ./cdc_state_checker ]; then
 	GO111MODULE=on go build
 fi

tests/cdc/cdc.go → tests/integration_tests/cdc/cdc.go

@@ -20,8 +20,8 @@ import (
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
-	"github.com/pingcap/ticdc/tests/dailytest"
-	"github.com/pingcap/ticdc/tests/util"
+	"github.com/pingcap/ticdc/tests/integration_tests/dailytest"
+	"github.com/pingcap/ticdc/tests/integration_tests/util"
 )
 
 func main() {

tests/dailytest/db.go → tests/integration_tests/dailytest/db.go

@@ -24,7 +24,7 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/parser/mysql"
-	"github.com/pingcap/ticdc/tests/util"
+	"github.com/pingcap/ticdc/tests/integration_tests/util"
 	"go.uber.org/zap/zapcore"
 )
 

tests/many_pk_or_uk/main.go → tests/integration_tests/many_pk_or_uk/main.go

@@ -23,7 +23,7 @@ import (
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
-	"github.com/pingcap/ticdc/tests/util"
+	"github.com/pingcap/ticdc/tests/integration_tests/util"
 )
 
 func main() {

tests/multi_source/main.go → tests/integration_tests/multi_source/main.go

@@ -28,7 +28,7 @@ import (
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
-	"github.com/pingcap/ticdc/tests/util"
+	"github.com/pingcap/ticdc/tests/integration_tests/util"
 	"go.uber.org/zap"
 )
 

tests/new_ci_collation/conf/diff_config.toml → tests/integration_tests/new_ci_collation_with_old_value/conf/diff_config.toml

@@ -13,7 +13,7 @@ check-struct-only = false
 
     target-instance = "tidb0"
 
-    target-check-tables = ["new_ci_collation_test.?*"]
+    target-check-tables = ["new_ci_collation_with_old_value_test.?*"]
 
 [data-sources]
 [data-sources.mysql1]

tests/new_ci_collation/data/test1.sql → tests/integration_tests/new_ci_collation_with_old_value/data/test1.sql

@@ -1,6 +1,6 @@
-drop database if exists `new_ci_collation_test`;
-create database `new_ci_collation_test`;
-use `new_ci_collation_test`;
+drop database if exists `new_ci_collation_with_old_value_test`;
+create database `new_ci_collation_with_old_value_test`;
+use `new_ci_collation_with_old_value_test`;
 
 CREATE TABLE t1 (
     a varchar(20) charset utf8mb4 collate utf8mb4_general_ci primary key,

tests/new_ci_collation/data/test2.sql → tests/integration_tests/new_ci_collation_with_old_value/data/test2.sql

@@ -1,4 +1,4 @@
-use `new_ci_collation_test`;
+use `new_ci_collation_with_old_value_test`;
 delete from t1;
 delete from t2;
 delete from t3;

tests/new_ci_collation/run.sh → tests/integration_tests/new_ci_collation_with_old_value/run.sh

@@ -20,7 +20,7 @@ function run() {
 
 	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
 
-	TOPIC_NAME="ticdc-new_ci_collation-test-$RANDOM"
+	TOPIC_NAME="ticdc-new_ci_collation_with_old_value-test-$RANDOM"
 	case $SINK_TYPE in
 	kafka) SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?partition-num=4&kafka-version=${KAFKA_VERSION}&max-message-bytes=10485760" ;;
 	*) SINK_URI="mysql://normal:123456@127.0.0.1:3306/?safe-mode=true" ;;
@@ -32,7 +32,7 @@ function run() {
 
 	run_sql_file $CUR/data/test1.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 	for i in $(seq 1 5); do
-		table="new_ci_collation_test.t$i"
+		table="new_ci_collation_with_old_value_test.t$i"
 		check_table_exists $table ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 	done
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml

tests/integration_tests/new_ci_collation_without_old_value/conf/changefeed.toml

@@ -0,0 +1 @@
+enable-old-value = false

tests/integration_tests/new_ci_collation_without_old_value/conf/diff_config.toml

@@ -0,0 +1,29 @@
+# diff Configuration.
+
+check-thread-count = 4
+
+export-fix-sql = true
+
+check-struct-only = false
+
+[task]
+output-dir = "/tmp/ticdc_dm_test/output"
+
+source-instances = ["mysql1"]
+
+target-instance = "tidb0"
+
+target-check-tables = ["new_ci_collation_without_old_value_test.?*"]
+
+[data-sources]
+[data-sources.mysql1]
+host = "127.0.0.1"
+port = 4000
+user = "root"
+password = ""
+
+[data-sources.tidb0]
+host = "127.0.0.1"
+port = 3306
+user = "root"
+password = ""

tests/integration_tests/new_ci_collation_without_old_value/conf/tidb_config.toml

@@ -0,0 +1,2 @@
+alter-primary-key = true
+new_collations_enabled_on_first_bootstrap = true

tests/integration_tests/new_ci_collation_without_old_value/data/test1.sql

@@ -0,0 +1,49 @@
+drop database if exists `new_ci_collation_without_old_value_test`;
+create database `new_ci_collation_without_old_value_test`;
+use `new_ci_collation_without_old_value_test`;
+
+CREATE TABLE t1 (
+    a varchar(20) charset utf8mb4 collate utf8mb4_general_ci primary key,
+    b int default 10
+);
+
+CREATE TABLE t2 (
+    a varchar(10) charset utf8 collate utf8_general_ci, primary key(a),
+    b int default 10
+);
+
+CREATE TABLE t3 (
+    id int primary key auto_increment,
+    a varchar(20) charset utf8mb4 collate utf8mb4_general_ci,
+    b int default 10
+);
+
+CREATE TABLE t4 (
+    a int primary key,
+    b varchar(10) charset utf8mb4 collate utf8mb4_general_ci,
+    c varchar(10) charset utf8 collate utf8_general_ci,
+    d varchar(10) not null,
+    unique key d(d)
+);
+
+CREATE TABLE t5 (
+    a varchar(10) charset utf8mb4 collate utf8mb4_bin, primary key(a),
+    b int default 10
+);
+
+
+insert into t1 (a) values ('A'),(' A'),('A\t'),('b'),('bA'),('bac'),('ab');
+insert into t1 (a) values ('😉');
+insert into t2 (a) values ('A'),(' A'),('A\t'),('b'),('bA'),('bac'),('ab');
+insert into t3 (a) values ('A'),('A '),('A   '),(' A'),('A\t'),('A\t ');
+insert into t3 (a) values ('a'),('a '),('a   '),(' a'),('a\t'),('a\t ');
+insert into t3 (a) values ('B'),('B '),('B   '),(' B'),('B\t'),('B\t ');
+insert into t3 (a) values ('b'),('b '),('b   '),(' b'),('b\t'),('b\t ');
+insert into t4 values (1,'A','A','1'),(2,'a\t','a\t','2'),(3,'ab','ab','3'),(4,'abc','abc','4');
+insert into t5 (a) values ('😉');
+insert into t5 (a) values ('a'),('A'),(' a'),(' A'),('a\t'),('ab'),('Ab');
+update t1 set b = b + 1;
+update t2 set b = 13;
+update t3 set b = 11 where a > 'A';
+drop index `primary` on t4;
+update t5 set b = 12;

tests/integration_tests/new_ci_collation_without_old_value/data/test2.sql

@@ -0,0 +1,6 @@
+use `new_ci_collation_without_old_value_test`;
+delete from t1;
+delete from t2;
+delete from t3;
+delete from t4;
+delete from t5;

tests/integration_tests/new_ci_collation_without_old_value/run.sh

@@ -0,0 +1,48 @@
+#!/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 run() {
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+
+	start_tidb_cluster --workdir $WORK_DIR --tidb-config $CUR/conf/tidb_config.toml
+
+	cd $WORK_DIR
+
+	# record tso before we create tables to skip the system table DDLs
+	start_ts=$(cdc cli tso query --pd=http://$UP_PD_HOST_1:$UP_PD_PORT_1)
+
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	TOPIC_NAME="ticdc-new_ci_collation_without_old_value-test-$RANDOM"
+	case $SINK_TYPE in
+	kafka) SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?partition-num=4&kafka-version=${KAFKA_VERSION}" ;;
+	*) SINK_URI="mysql://normal:123456@127.0.0.1:3306/?safe-mode=true" ;;
+	esac
+	cdc cli changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" --config $CUR/conf/changefeed.toml
+	if [ "$SINK_TYPE" == "kafka" ]; then
+		run_kafka_consumer $WORK_DIR "kafka://127.0.0.1:9092/$TOPIC_NAME?partition-num=4&version=${KAFKA_VERSION}"
+	fi
+
+	run_sql_file $CUR/data/test1.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	for i in $(seq 1 5); do
+		table="new_ci_collation_without_old_value_test.t$i"
+		check_table_exists $table ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	done
+	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
+	run_sql_file $CUR/data/test2.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
+
+	cleanup_process $CDC_BINARY
+}
+
+trap stop_tidb_cluster EXIT
+run $*
+check_logs $WORK_DIR
+echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"