tests(ticdc): add storage integration tests (#8685)

close #8826

tests/integration_tests/api_v2/run.sh

@@ -9,10 +9,6 @@ CDC_BINARY=cdc.test
 SINK_TYPE=$1
 
 function prepare() {
-	if [ "$SINK_TYPE" == "kafka" ]; then
-		return
-	fi
-
 	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
 
 	start_tidb_cluster --workdir $WORK_DIR
@@ -29,8 +25,8 @@ function prepare() {
 }
 
 trap stop_tidb_cluster EXIT
-# kafka is not supported yet.
-if [ "$SINK_TYPE" != "kafka" ]; then
+# kafka and storage is not supported yet.
+if [ "$SINK_TYPE" == "mysql" ]; then
 	prepare $*
 
 	cd "$(dirname "$0")"

tests/integration_tests/bank/run.sh

@@ -9,10 +9,6 @@ CDC_BINARY=cdc.test
 SINK_TYPE=$1
 
 function prepare() {
-	if [ "$SINK_TYPE" == "kafka" ]; then
-		return
-	fi
-
 	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
 
 	start_tidb_cluster --workdir $WORK_DIR
@@ -30,7 +26,7 @@ function prepare() {
 
 trap stop_tidb_cluster EXIT
 # kafka is not supported yet.
-if [ "$SINK_TYPE" != "kafka" ]; then
+if [ "$SINK_TYPE" == "mysql" ]; then
 	prepare $*
 
 	cd "$(dirname "$0")"

tests/integration_tests/batch_update_to_no_batch/run.sh

@@ -14,7 +14,7 @@ SINK_TYPE=$1
 # 3. cdc can switch from batch mode to no-batch mode and vice versa and works well
 function run() {
 	# batch mode only supports mysql sink
-	if [ "$SINK_TYPE" == "kafka" ]; then
+	if [ "$SINK_TYPE" != "mysql" ]; then
 		return
 	fi
 

tests/integration_tests/bdr_mode/run.sh

@@ -10,7 +10,7 @@ SINK_TYPE=$1
 
 function run() {
 	# BDR mode only supports mysql sink
-	if [ "$SINK_TYPE" == "kafka" ]; then
+	if [ "$SINK_TYPE" != "mysql" ]; then
 		return
 	fi
 

tests/integration_tests/canal_json_adapter_compatibility/run.sh

@@ -11,7 +11,7 @@ SINK_TYPE=$1
 # use canal-adapter to sync data from kafka to mysql,
 # make sure that `canal-json` output can be consumed by the canal-adapter.
 function run() {
-	if [ "$SINK_TYPE" == "mysql" ]; then
+	if [ "$SINK_TYPE" != "kafka" ]; then
 		return
 	fi
 

tests/integration_tests/canal_json_basic/run.sh

@@ -10,7 +10,7 @@ SINK_TYPE=$1
 
 # use kafka-consumer with canal-json decoder to sync data from kafka to mysql
 function run() {
-	if [ "$SINK_TYPE" == "mysql" ]; then
+	if [ "$SINK_TYPE" != "kafka" ]; then
 		return
 	fi
 

tests/integration_tests/canal_json_storage_basic/run.sh

@@ -9,9 +9,7 @@ CDC_BINARY=cdc.test
 SINK_TYPE=$1
 
 function run() {
-	# Now, we run the storage tests in mysql sink tests.
-	# It's a temporary solution, we will move it to a new test pipeline later.
-	if [ "$SINK_TYPE" != "mysql" ]; then
+	if [ "$SINK_TYPE" != "storage" ]; then
 		return
 	fi
 

tests/integration_tests/canal_json_storage_partition_table/run.sh

@@ -9,9 +9,7 @@ CDC_BINARY=cdc.test
 SINK_TYPE=$1
 
 function run() {
-	# Now, we run the storage tests in mysql sink tests.
-	# It's a temporary solution, we will move it to a new test pipeline later.
-	if [ "$SINK_TYPE" != "mysql" ]; then
+	if [ "$SINK_TYPE" != "storage" ]; then
 		return
 	fi
 

tests/integration_tests/cdc/cdc.go

@@ -20,7 +20,7 @@ import (
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
-	"github.com/pingcap/tiflow/tests/integration_tests/dailytest"
+	"github.com/pingcap/tiflow/tests/integration_tests/cdc/dailytest"
 	"github.com/pingcap/tiflow/tests/integration_tests/util"
 )
 

tests/integration_tests/changefeed_fast_fail/run.sh

@@ -12,7 +12,7 @@ MAX_RETRIES=20
 function run() {
 	# it is no need to test kafka
 	# the logic are all the same
-	if [ "$SINK_TYPE" == "kafka" ]; then
+	if [ "$SINK_TYPE" != "mysql" ]; then
 		return
 	fi
 

tests/integration_tests/changefeed_resume_with_checkpoint_ts/run.sh

@@ -10,11 +10,6 @@ SINK_TYPE=$1
 MAX_RETRIES=20
 
 function prepare() {
-	if [ "$SINK_TYPE" == "kafka" ]; then
-		echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"
-		exit 0
-	fi
-
 	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
 	start_tidb_cluster --workdir $WORK_DIR
 	cd $WORK_DIR
@@ -132,9 +127,11 @@ function resume_changefeed_in_failed_state() {
 }
 
 trap stop_tidb_cluster EXIT
-prepare
-resume_changefeed_in_stopped_state $*
-resume_changefeed_in_failed_state $*
 
-check_logs $WORK_DIR
+if [ "$SINK_TYPE" == "mysql" ]; then
+	prepare
+	resume_changefeed_in_stopped_state $*
+	resume_changefeed_in_failed_state $*
+	check_logs $WORK_DIR
+fi
 echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"

tests/integration_tests/clustered_index/run.sh

@@ -46,8 +46,8 @@ function run() {
 
 # kafka is not supported yet.
 # ref to issue: https://github.com/pingcap/tiflow/issues/3421
-if [ "$SINK_TYPE" = "kafka" ]; then
-	echo "[$(date)] <<<<<< skip test case $TEST_NAME for kafka! >>>>>>"
+if [ "$SINK_TYPE" != "mysql" ]; then
+	echo "[$(date)] <<<<<< skip test case $TEST_NAME for $SINK_TYPE! >>>>>>"
 	exit 0
 fi
 trap stop_tidb_cluster EXIT

tests/integration_tests/consistent_replicate_ddl/run.sh

@@ -19,7 +19,7 @@ stop() {
 
 function run() {
 	# we only support eventually consistent replication with MySQL sink
-	if [ "$SINK_TYPE" == "kafka" ]; then
+	if [ "$SINK_TYPE" != "mysql" ]; then
 		return
 	fi
 

tests/integration_tests/consistent_replicate_gbk/run.sh

@@ -43,7 +43,7 @@ s3cmd --access_key=$MINIO_ACCESS_KEY --secret_key=$MINIO_SECRET_KEY --host=$S3_E
 
 function run() {
 	# we only support eventually consistent replication with MySQL sink
-	if [ "$SINK_TYPE" == "kafka" ]; then
+	if [ "$SINK_TYPE" != "mysql" ]; then
 		return
 	fi
 

tests/integration_tests/consistent_replicate_nfs/run.sh

@@ -16,7 +16,7 @@ stop() {
 
 function run() {
 	# we only support eventually consistent replication with MySQL sink
-	if [ "$SINK_TYPE" == "kafka" ]; then
+	if [ "$SINK_TYPE" != "mysql" ]; then
 		return
 	fi
 

tests/integration_tests/consistent_replicate_storage_file/run.sh

@@ -19,7 +19,7 @@ stop() {
 
 function run() {
 	# we only support eventually consistent replication with MySQL sink
-	if [ "$SINK_TYPE" == "kafka" ]; then
+	if [ "$SINK_TYPE" != "mysql" ]; then
 		return
 	fi
 

tests/integration_tests/consistent_replicate_storage_s3/run.sh

@@ -45,7 +45,7 @@ s3cmd --access_key=$MINIO_ACCESS_KEY --secret_key=$MINIO_SECRET_KEY --host=$S3_E
 
 function run() {
 	# we only support eventually consistent replication with MySQL sink
-	if [ "$SINK_TYPE" == "kafka" ]; then
+	if [ "$SINK_TYPE" != "mysql" ]; then
 		return
 	fi
 

tests/integration_tests/csv_storage_basic/run.sh

@@ -9,9 +9,7 @@ CDC_BINARY=cdc.test
 SINK_TYPE=$1
 
 function run() {
-	# Now, we run the storage tests in mysql sink tests.
-	# It's a temporary solution, we will move it to a new test pipeline later.
-	if [ "$SINK_TYPE" != "mysql" ]; then
+	if [ "$SINK_TYPE" != "storage" ]; then
 		return
 	fi
 

tests/integration_tests/csv_storage_multi_tables_ddl/run.sh

@@ -11,7 +11,7 @@ SINK_TYPE=$1
 function run() {
 	# Now, we run the storage tests in mysql sink tests.
 	# It's a temporary solution, we will move it to a new test pipeline later.
-	if [ "$SINK_TYPE" != "mysql" ]; then
+	if [ "$SINK_TYPE" != "storage" ]; then
 		return
 	fi
 

tests/integration_tests/csv_storage_partition_table/run.sh

@@ -9,9 +9,7 @@ CDC_BINARY=cdc.test
 SINK_TYPE=$1
 
 function run() {
-	# Now, we run the storage tests in mysql sink tests.
-	# It's a temporary solution, we will move it to a new test pipeline later.
-	if [ "$SINK_TYPE" != "mysql" ]; then
+	if [ "$SINK_TYPE" != "storage" ]; then
 		return
 	fi
 

tests/integration_tests/ddl_reentrant/run.sh

@@ -119,7 +119,7 @@ function ddl_test() {
 
 function run() {
 	# don't test kafka in this case
-	if [ "$SINK_TYPE" == "kafka" ]; then
+	if [ "$SINK_TYPE" != "mysql" ]; then
 		return
 	fi
 

tests/integration_tests/http_api/run.sh

@@ -10,8 +10,8 @@ SINK_TYPE=$1
 MAX_RETRIES=50
 
 function run() {
-	# mysql and kafka are the same
-	if [ "$SINK_TYPE" == "kafka" ]; then
+	# storage and kafka are the same as mysql
+	if [ "$SINK_TYPE" != "mysql" ]; then
 		return
 	fi
 

tests/integration_tests/http_api_tls/run.sh

@@ -11,8 +11,8 @@ TLS_DIR=$(cd $CUR/../_certificates && pwd)
 MAX_RETRIES=20
 
 function run() {
-	# mysql and kafka are the same
-	if [ "$SINK_TYPE" == "kafka" ]; then
+	# storage and kafka are the same as mysql
+	if [ "$SINK_TYPE" != "mysql" ]; then
 		return
 	fi
 

tests/integration_tests/kafka_big_messages/run.sh

@@ -10,7 +10,7 @@ SINK_TYPE=$1
 
 function run() {
 	# test kafka sink only in this case
-	if [ "$SINK_TYPE" == "mysql" ]; then
+	if [ "$SINK_TYPE" != "kafka" ]; then
 		return
 	fi
 	rm -rf $WORK_DIR && mkdir -p $WORK_DIR

tests/integration_tests/kafka_big_messages_v2/run.sh

@@ -10,7 +10,7 @@ SINK_TYPE=$1
 
 function run() {
 	# test kafka sink only in this case
-	if [ "$SINK_TYPE" == "mysql" ]; then
+	if [ "$SINK_TYPE" != "kafka" ]; then
 		return
 	fi
 	rm -rf $WORK_DIR && mkdir -p $WORK_DIR

tests/integration_tests/big_txn_v2/conf/diff_config.toml → tests/integration_tests/kafka_big_txn_v2/conf/diff_config.toml

@@ -7,7 +7,7 @@ export-fix-sql = true
 check-struct-only = false
 
 [task]
-    output-dir = "/tmp/tidb_cdc_test/big_txn/sync_diff/output"
+    output-dir = "/tmp/tidb_cdc_test/kafka_big_txn_v2/sync_diff/output"
 
     source-instances = ["tidb"]
 

tests/integration_tests/big_txn_v2/run.sh → tests/integration_tests/kafka_big_txn_v2/run.sh

@@ -12,7 +12,8 @@ CDC_COUNT=3
 DB_COUNT=4
 
 function run() {
-	if [ "$SINK_TYPE" == "mysql" ]; then
+	# test kafka sink only in this case
+	if [ "$SINK_TYPE" != "kafka" ]; then
 		return
 	fi
 	rm -rf $WORK_DIR && mkdir -p $WORK_DIR

tests/integration_tests/kafka_compression/run.sh

@@ -30,7 +30,7 @@ function test_compression() {
 }
 
 function run() {
-	if [ "$SINK_TYPE" == "mysql" ]; then
+	if [ "$SINK_TYPE" != "kafka" ]; then
 		return
 	fi
 

tests/integration_tests/kafka_messages/run.sh

@@ -12,11 +12,6 @@ CDC_COUNT=3
 DB_COUNT=4
 
 function run_length_limit() {
-	# test kafka sink only in this case
-	if [ "$SINK_TYPE" == "mysql" ]; then
-		return
-	fi
-
 	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
 
 	start_tidb_cluster --workdir $WORK_DIR
@@ -71,11 +66,6 @@ function run_length_limit() {
 }
 
 function run_batch_size_limit() {
-	# test kafka sink only in this case
-	if [ "$SINK_TYPE" == "mysql" ]; then
-		return
-	fi
-
 	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
 
 	start_tidb_cluster --workdir $WORK_DIR
@@ -126,8 +116,16 @@ function run_batch_size_limit() {
 	stop_tidb_cluster
 }
 
+function run() {
+	# test kafka sink only in this case
+	if [ "$SINK_TYPE" != "kafka" ]; then
+		return
+	fi
+
+	run_length_limit $*
+	run_batch_size_limit $*
+}
+
 trap stop_tidb_cluster EXIT
-run_length_limit $*
-run_batch_size_limit $*
 check_logs $WORK_DIR
 echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"

tests/integration_tests/kafka_sink_error_resume/run.sh

@@ -14,7 +14,7 @@ MAX_RETRIES=20
 
 function run() {
 	# test kafka sink only in this case
-	if [ "$SINK_TYPE" == "mysql" ]; then
+	if [ "$SINK_TYPE" != "kafka" ]; then
 		return
 	fi
 

tests/integration_tests/kill_owner_with_ddl/run.sh

@@ -39,7 +39,7 @@ export -f kill_cdc_and_restart
 
 function run() {
 	# kafka is not supported yet.
-	if [ "$SINK_TYPE" == "kafka" ]; then
+	if [ "$SINK_TYPE" != "mysql" ]; then
 		return
 	fi
 

tests/integration_tests/multi_cdc_cluster/run.sh

@@ -10,7 +10,7 @@ SINK_TYPE=$1
 
 function run() {
 	# test mysql sink only in this case
-	if [ "$SINK_TYPE" == "kafka" ]; then
+	if [ "$SINK_TYPE" != "mysql" ]; then
 		return
 	fi
 	rm -rf $WORK_DIR && mkdir -p $WORK_DIR

tests/integration_tests/multi_changefeed/run.sh

@@ -57,7 +57,7 @@ export -f check_old_value_enabled
 
 function run() {
 	# kafka is not supported yet.
-	if [ "$SINK_TYPE" == "kafka" ]; then
+	if [ "$SINK_TYPE" != "mysql" ]; then
 		return
 	fi
 

tests/integration_tests/multi_tables_ddl_v2/run.sh

@@ -9,7 +9,8 @@ CDC_BINARY=cdc.test
 SINK_TYPE=$1
 
 function run() {
-	if [ "$SINK_TYPE" == "mysql" ]; then
+	# test kafka sink only in this case
+	if [ "$SINK_TYPE" != "kafka" ]; then
 		return
 	fi