diff --git a/e2e_test/source_legacy/basic/scripts/README.md b/e2e_test/source_legacy/basic/scripts/README.md
new file mode 100644
index 0000000000000..928394a43007e
--- /dev/null
+++ b/e2e_test/source_legacy/basic/scripts/README.md
@@ -0,0 +1,12 @@
+This folder contains scripts to prepare data for testing sources.
+
+## Kafka
+
+`e2e_test/source_legacy/basic/scripts/test_data` contains the data. Filename's convention is `<topic_name>.<n_partitions>`.
+
+- If `<topic_name>` ends with `bin`, the whole file is a message with binary data.
+- If `<topic_name>` ends with `avro_json` or `json_schema`:
+    - The first line is the schema. Key and value are separated by `^`.
+    - The rest of the lines are messages in JSON format. Key and value are separated by `^`.
+    - Produced to Kafka with `schema_registry_producer.py` (serialized to Avro or JSON)
+- Otherwise, each line is a message, and key/value is separated by `^`.
diff --git a/scripts/source/prepare_ci_kafka.sh b/e2e_test/source_legacy/basic/scripts/prepare_ci_kafka.sh
similarity index 91%
rename from scripts/source/prepare_ci_kafka.sh
rename to e2e_test/source_legacy/basic/scripts/prepare_ci_kafka.sh
index 27c88e6e63af2..34d40646861a5 100755
--- a/scripts/source/prepare_ci_kafka.sh
+++ b/e2e_test/source_legacy/basic/scripts/prepare_ci_kafka.sh
@@ -4,16 +4,16 @@
 set -e
 
 SCRIPT_PATH="$(cd "$(dirname "$0")" >/dev/null 2>&1 && pwd)"
-cd "$SCRIPT_PATH/.." || exit 1
-# cwd is /scripts
+# SCRIPT_PATH is e2e_test/source_legacy/kafka/script/
+# cwd is e2e_test/source_legacy/kafka/
 
 echo "$SCRIPT_PATH"
 
-source ../.risingwave/config/risedev-env
+source ../../../.risingwave/config/risedev-env
 
 if [ "$1" == "compress" ]; then
   echo "Compress test_data/ into test_data.zip"
-  cd ./source
+  cd ./script
   zip_file=test_data.zip
   if [ -f "$zip_file" ]; then
     rm "$zip_file"
@@ -23,7 +23,7 @@ if [ "$1" == "compress" ]; then
 fi
 
 echo "--- Extract data for Kafka"
-cd ./source/
+cd ./script/
 mkdir -p ./test_data/ch_benchmark/
 unzip -o test_data.zip -d .
 cd ..
diff --git a/scripts/source/schema_registry_producer.py b/e2e_test/source_legacy/basic/scripts/schema_registry_producer.py
similarity index 100%
rename from scripts/source/schema_registry_producer.py
rename to e2e_test/source_legacy/basic/scripts/schema_registry_producer.py
diff --git a/scripts/source/test_data.zip b/e2e_test/source_legacy/basic/scripts/test_data.zip
similarity index 100%
rename from scripts/source/test_data.zip
rename to e2e_test/source_legacy/basic/scripts/test_data.zip
diff --git a/scripts/source/test_data/avro_complex_schema_bin.1 b/e2e_test/source_legacy/basic/scripts/test_data/avro_complex_schema_bin.1
similarity index 100%
rename from scripts/source/test_data/avro_complex_schema_bin.1
rename to e2e_test/source_legacy/basic/scripts/test_data/avro_complex_schema_bin.1
diff --git a/scripts/source/test_data/avro_simple_schema_bin.1 b/e2e_test/source_legacy/basic/scripts/test_data/avro_simple_schema_bin.1
similarity index 100%
rename from scripts/source/test_data/avro_simple_schema_bin.1
rename to e2e_test/source_legacy/basic/scripts/test_data/avro_simple_schema_bin.1
diff --git a/scripts/source/test_data/bug_bash.1 b/e2e_test/source_legacy/basic/scripts/test_data/bug_bash.1
similarity index 100%
rename from scripts/source/test_data/bug_bash.1
rename to e2e_test/source_legacy/basic/scripts/test_data/bug_bash.1
diff --git a/scripts/source/test_data/canal_json_double_field.1 b/e2e_test/source_legacy/basic/scripts/test_data/canal_json_double_field.1
similarity index 100%
rename from scripts/source/test_data/canal_json_double_field.1
rename to e2e_test/source_legacy/basic/scripts/test_data/canal_json_double_field.1
diff --git a/scripts/source/test_data/cannal_json.1 b/e2e_test/source_legacy/basic/scripts/test_data/cannal_json.1
similarity index 100%
rename from scripts/source/test_data/cannal_json.1
rename to e2e_test/source_legacy/basic/scripts/test_data/cannal_json.1
diff --git a/scripts/source/test_data/debezium_compact_avro_json.1 b/e2e_test/source_legacy/basic/scripts/test_data/debezium_compact_avro_json.1
similarity index 100%
rename from scripts/source/test_data/debezium_compact_avro_json.1
rename to e2e_test/source_legacy/basic/scripts/test_data/debezium_compact_avro_json.1
diff --git a/scripts/source/test_data/debezium_ignore_case_json.1 b/e2e_test/source_legacy/basic/scripts/test_data/debezium_ignore_case_json.1
similarity index 100%
rename from scripts/source/test_data/debezium_ignore_case_json.1
rename to e2e_test/source_legacy/basic/scripts/test_data/debezium_ignore_case_json.1
diff --git a/scripts/source/test_data/debezium_log.1 b/e2e_test/source_legacy/basic/scripts/test_data/debezium_log.1
similarity index 100%
rename from scripts/source/test_data/debezium_log.1
rename to e2e_test/source_legacy/basic/scripts/test_data/debezium_log.1
diff --git a/scripts/source/test_data/debezium_log_no_schema_field.1 b/e2e_test/source_legacy/basic/scripts/test_data/debezium_log_no_schema_field.1
similarity index 100%
rename from scripts/source/test_data/debezium_log_no_schema_field.1
rename to e2e_test/source_legacy/basic/scripts/test_data/debezium_log_no_schema_field.1
diff --git a/scripts/source/test_data/debezium_mess_key.1 b/e2e_test/source_legacy/basic/scripts/test_data/debezium_mess_key.1
similarity index 100%
rename from scripts/source/test_data/debezium_mess_key.1
rename to e2e_test/source_legacy/basic/scripts/test_data/debezium_mess_key.1
diff --git a/scripts/source/test_data/debezium_mongo_json_customers.1 b/e2e_test/source_legacy/basic/scripts/test_data/debezium_mongo_json_customers.1
similarity index 100%
rename from scripts/source/test_data/debezium_mongo_json_customers.1
rename to e2e_test/source_legacy/basic/scripts/test_data/debezium_mongo_json_customers.1
diff --git a/scripts/source/test_data/debezium_mongo_json_customers_no_schema_field.1 b/e2e_test/source_legacy/basic/scripts/test_data/debezium_mongo_json_customers_no_schema_field.1
similarity index 100%
rename from scripts/source/test_data/debezium_mongo_json_customers_no_schema_field.1
rename to e2e_test/source_legacy/basic/scripts/test_data/debezium_mongo_json_customers_no_schema_field.1
diff --git a/scripts/source/test_data/debezium_non_compact_avro_json.1 b/e2e_test/source_legacy/basic/scripts/test_data/debezium_non_compact_avro_json.1
similarity index 100%
rename from scripts/source/test_data/debezium_non_compact_avro_json.1
rename to e2e_test/source_legacy/basic/scripts/test_data/debezium_non_compact_avro_json.1
diff --git a/scripts/source/test_data/json_bytea.1 b/e2e_test/source_legacy/basic/scripts/test_data/json_bytea.1
similarity index 100%
rename from scripts/source/test_data/json_bytea.1
rename to e2e_test/source_legacy/basic/scripts/test_data/json_bytea.1
diff --git a/scripts/source/test_data/json_c.1 b/e2e_test/source_legacy/basic/scripts/test_data/json_c.1
similarity index 100%
rename from scripts/source/test_data/json_c.1
rename to e2e_test/source_legacy/basic/scripts/test_data/json_c.1
diff --git a/scripts/source/test_data/json_schema_without_schema.1 b/e2e_test/source_legacy/basic/scripts/test_data/json_schema_without_schema.1
similarity index 100%
rename from scripts/source/test_data/json_schema_without_schema.1
rename to e2e_test/source_legacy/basic/scripts/test_data/json_schema_without_schema.1
diff --git a/scripts/source/test_data/json_timestamptz_handling_mode.1 b/e2e_test/source_legacy/basic/scripts/test_data/json_timestamptz_handling_mode.1
similarity index 100%
rename from scripts/source/test_data/json_timestamptz_handling_mode.1
rename to e2e_test/source_legacy/basic/scripts/test_data/json_timestamptz_handling_mode.1
diff --git a/scripts/source/test_data/kafka_1_csv_topic.1 b/e2e_test/source_legacy/basic/scripts/test_data/kafka_1_csv_topic.1
similarity index 100%
rename from scripts/source/test_data/kafka_1_csv_topic.1
rename to e2e_test/source_legacy/basic/scripts/test_data/kafka_1_csv_topic.1
diff --git a/scripts/source/test_data/kafka_1_partition_mv_topic.1 b/e2e_test/source_legacy/basic/scripts/test_data/kafka_1_partition_mv_topic.1
similarity index 100%
rename from scripts/source/test_data/kafka_1_partition_mv_topic.1
rename to e2e_test/source_legacy/basic/scripts/test_data/kafka_1_partition_mv_topic.1
diff --git a/scripts/source/test_data/kafka_1_partition_topic.1 b/e2e_test/source_legacy/basic/scripts/test_data/kafka_1_partition_topic.1
similarity index 100%
rename from scripts/source/test_data/kafka_1_partition_topic.1
rename to e2e_test/source_legacy/basic/scripts/test_data/kafka_1_partition_topic.1
diff --git a/scripts/source/test_data/kafka_2_partition_topic.2 b/e2e_test/source_legacy/basic/scripts/test_data/kafka_2_partition_topic.2
similarity index 100%
rename from scripts/source/test_data/kafka_2_partition_topic.2
rename to e2e_test/source_legacy/basic/scripts/test_data/kafka_2_partition_topic.2
diff --git a/scripts/source/test_data/kafka_3_partition_topic.3 b/e2e_test/source_legacy/basic/scripts/test_data/kafka_3_partition_topic.3
similarity index 100%
rename from scripts/source/test_data/kafka_3_partition_topic.3
rename to e2e_test/source_legacy/basic/scripts/test_data/kafka_3_partition_topic.3
diff --git a/scripts/source/test_data/kafka_4_partition_topic.4 b/e2e_test/source_legacy/basic/scripts/test_data/kafka_4_partition_topic.4
similarity index 100%
rename from scripts/source/test_data/kafka_4_partition_topic.4
rename to e2e_test/source_legacy/basic/scripts/test_data/kafka_4_partition_topic.4
diff --git a/scripts/source/test_data/kafka_4_partition_topic_generated_columns.1 b/e2e_test/source_legacy/basic/scripts/test_data/kafka_4_partition_topic_generated_columns.1
similarity index 100%
rename from scripts/source/test_data/kafka_4_partition_topic_generated_columns.1
rename to e2e_test/source_legacy/basic/scripts/test_data/kafka_4_partition_topic_generated_columns.1
diff --git a/scripts/source/test_data/kafka_4_partition_topic_with_100_message.4 b/e2e_test/source_legacy/basic/scripts/test_data/kafka_4_partition_topic_with_100_message.4
similarity index 100%
rename from scripts/source/test_data/kafka_4_partition_topic_with_100_message.4
rename to e2e_test/source_legacy/basic/scripts/test_data/kafka_4_partition_topic_with_100_message.4
diff --git a/scripts/source/test_data/kafka_json_schema.1 b/e2e_test/source_legacy/basic/scripts/test_data/kafka_json_schema.1
similarity index 100%
rename from scripts/source/test_data/kafka_json_schema.1
rename to e2e_test/source_legacy/basic/scripts/test_data/kafka_json_schema.1
diff --git a/scripts/source/test_data/kafka_source_format_bytes.1 b/e2e_test/source_legacy/basic/scripts/test_data/kafka_source_format_bytes.1
similarity index 100%
rename from scripts/source/test_data/kafka_source_format_bytes.1
rename to e2e_test/source_legacy/basic/scripts/test_data/kafka_source_format_bytes.1
diff --git a/scripts/source/test_data/kafka_upsert_json_schema.1 b/e2e_test/source_legacy/basic/scripts/test_data/kafka_upsert_json_schema.1
similarity index 100%
rename from scripts/source/test_data/kafka_upsert_json_schema.1
rename to e2e_test/source_legacy/basic/scripts/test_data/kafka_upsert_json_schema.1
diff --git a/scripts/source/test_data/maxwell_json.1 b/e2e_test/source_legacy/basic/scripts/test_data/maxwell_json.1
similarity index 100%
rename from scripts/source/test_data/maxwell_json.1
rename to e2e_test/source_legacy/basic/scripts/test_data/maxwell_json.1
diff --git a/scripts/source/test_data/proto_c_bin.1 b/e2e_test/source_legacy/basic/scripts/test_data/proto_c_bin.1
similarity index 100%
rename from scripts/source/test_data/proto_c_bin.1
rename to e2e_test/source_legacy/basic/scripts/test_data/proto_c_bin.1
diff --git a/scripts/source/test_data/sink_debezium.1 b/e2e_test/source_legacy/basic/scripts/test_data/sink_debezium.1
similarity index 100%
rename from scripts/source/test_data/sink_debezium.1
rename to e2e_test/source_legacy/basic/scripts/test_data/sink_debezium.1
diff --git a/scripts/source/test_data/sink_target.1 b/e2e_test/source_legacy/basic/scripts/test_data/sink_target.1
similarity index 100%
rename from scripts/source/test_data/sink_target.1
rename to e2e_test/source_legacy/basic/scripts/test_data/sink_target.1
diff --git a/scripts/source/test_data/upsert_json.1 b/e2e_test/source_legacy/basic/scripts/test_data/upsert_json.1
similarity index 100%
rename from scripts/source/test_data/upsert_json.1
rename to e2e_test/source_legacy/basic/scripts/test_data/upsert_json.1
diff --git a/scripts/source/test_data/upsert_student_avro_json.1 b/e2e_test/source_legacy/basic/scripts/test_data/upsert_student_avro_json.1
similarity index 100%
rename from scripts/source/test_data/upsert_student_avro_json.1
rename to e2e_test/source_legacy/basic/scripts/test_data/upsert_student_avro_json.1
diff --git a/scripts/source/test_data/upsert_student_key_not_subset_of_value_avro_json.1 b/e2e_test/source_legacy/basic/scripts/test_data/upsert_student_key_not_subset_of_value_avro_json.1
similarity index 100%
rename from scripts/source/test_data/upsert_student_key_not_subset_of_value_avro_json.1
rename to e2e_test/source_legacy/basic/scripts/test_data/upsert_student_key_not_subset_of_value_avro_json.1
diff --git a/scripts/source/test_data/weiling.1 b/e2e_test/source_legacy/basic/scripts/test_data/weiling.1
similarity index 100%
rename from scripts/source/test_data/weiling.1
rename to e2e_test/source_legacy/basic/scripts/test_data/weiling.1
diff --git a/scripts/source/README.md b/scripts/source/README.md
index 311c058a8230b..928394a43007e 100644
--- a/scripts/source/README.md
+++ b/scripts/source/README.md
@@ -2,7 +2,7 @@ This folder contains scripts to prepare data for testing sources.
 
 ## Kafka
 
-`scripts/source/test_data` contains the data. Filename's convention is `<topic_name>.<n_partitions>`.
+`e2e_test/source_legacy/basic/scripts/test_data` contains the data. Filename's convention is `<topic_name>.<n_partitions>`.
 
 - If `<topic_name>` ends with `bin`, the whole file is a message with binary data.
 - If `<topic_name>` ends with `avro_json` or `json_schema`: