fix ci failure

Signed-off-by: BornChanger <dawn_catcher@126.com>

Author: BornChanger

.gitignore

@@ -18,3 +18,6 @@ cdc-junit-report.xml
 # temp files
 *.tmp
 metastore_db/
+
+.gocache/
+.gopath/

tests/integration_tests/iceberg_append_basic/run.sh

@@ -37,7 +37,8 @@ function prepare() {
 
 	WAREHOUSE_DIR="$WORK_DIR/iceberg_warehouse"
 	SINK_URI="iceberg://?warehouse=file://$WAREHOUSE_DIR&catalog=hadoop&namespace=ns&mode=append&commit-interval=1s&enable-checkpoint-table=true&enable-global-checkpoint-table=true&partitioning=days(_tidb_commit_time)"
-	cdc_cli_changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI"
+	changefeed_id=$(cdc_cli_changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" | grep '^ID:' | head -n1 | awk '{print $2}')
+	wait_changefeed_table_assigned "$changefeed_id" "test" "iceberg_append_basic"
 }
 
 function wait_file_exists() {
@@ -55,6 +56,34 @@ function wait_file_exists() {
 	return 1
 }
 
+function wait_changefeed_table_assigned() {
+	changefeed_id=$1
+	db_name=$2
+	table_name=$3
+
+	if [ "${TICDC_NEWARCH:-}" != "true" ]; then
+		return 0
+	fi
+
+	table_id=$(get_table_id "$db_name" "$table_name")
+	echo "wait table ${db_name}.${table_name} (id=${table_id}) assigned to changefeed ${changefeed_id}"
+
+	i=0
+	while [ $i -lt 60 ]; do
+		if curl -s "http://127.0.0.1:8300/api/v2/changefeeds/${changefeed_id}/tables?keyspace=${KEYSPACE_NAME}" \
+			| jq -e --argjson tid "$table_id" \
+				'def items: (if type=="array" then . else .items // [] end); items[]?.table_ids[]? | select(. == $tid)' >/dev/null; then
+			return 0
+		fi
+		sleep 2
+		((i++))
+	done
+
+	echo "table ${db_name}.${table_name} not assigned to changefeed ${changefeed_id} after $((i * 2))s"
+	curl -s "http://127.0.0.1:8300/api/v2/changefeeds/${changefeed_id}/tables?keyspace=${KEYSPACE_NAME}" | jq . || true
+	return 1
+}
+
 function iceberg_check_append_basic() {
 	do_retry 5 2 run_sql "INSERT INTO test.iceberg_append_basic(id, val) VALUES (1, 1);"
 	do_retry 5 2 run_sql "INSERT INTO test.iceberg_append_basic(id, val) VALUES (2, 2);"

tests/integration_tests/iceberg_upsert_basic/run.sh

@@ -37,7 +37,8 @@ function prepare() {
 
 	WAREHOUSE_DIR="$WORK_DIR/iceberg_warehouse"
 	SINK_URI="iceberg://?warehouse=file://$WAREHOUSE_DIR&catalog=hadoop&namespace=ns&mode=upsert&commit-interval=1s&enable-checkpoint-table=true&partitioning=none"
-	cdc_cli_changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI"
+	changefeed_id=$(cdc_cli_changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" | grep '^ID:' | head -n1 | awk '{print $2}')
+	wait_changefeed_table_assigned "$changefeed_id" "test" "iceberg_upsert_basic"
 }
 
 function wait_file_exists() {
@@ -55,6 +56,34 @@ function wait_file_exists() {
 	return 1
 }
 
+function wait_changefeed_table_assigned() {
+	changefeed_id=$1
+	db_name=$2
+	table_name=$3
+
+	if [ "${TICDC_NEWARCH:-}" != "true" ]; then
+		return 0
+	fi
+
+	table_id=$(get_table_id "$db_name" "$table_name")
+	echo "wait table ${db_name}.${table_name} (id=${table_id}) assigned to changefeed ${changefeed_id}"
+
+	i=0
+	while [ $i -lt 60 ]; do
+		if curl -s "http://127.0.0.1:8300/api/v2/changefeeds/${changefeed_id}/tables?keyspace=${KEYSPACE_NAME}" \
+			| jq -e --argjson tid "$table_id" \
+				'def items: (if type=="array" then . else .items // [] end); items[]?.table_ids[]? | select(. == $tid)' >/dev/null; then
+			return 0
+		fi
+		sleep 2
+		((i++))
+	done
+
+	echo "table ${db_name}.${table_name} not assigned to changefeed ${changefeed_id} after $((i * 2))s"
+	curl -s "http://127.0.0.1:8300/api/v2/changefeeds/${changefeed_id}/tables?keyspace=${KEYSPACE_NAME}" | jq . || true
+	return 1
+}
+
 function iceberg_check_upsert_basic() {
 	do_retry 5 2 run_sql "INSERT INTO test.iceberg_upsert_basic(id, val) VALUES (1, 1);"
 	do_retry 5 2 run_sql "INSERT INTO test.iceberg_upsert_basic(id, val) VALUES (2, 2);"