Merge branch 'main' into xxchan/test

Makefile.toml

@@ -1317,7 +1317,6 @@ dependencies = ["k", "l", "check-logs", "wait-processes-exit"]
 description = "Kill cluster, dump logs and check logs"
 
 [tasks.wait-processes-exit]
-private = true
 category = "Misc"
 description = "Wait for RisingWave processes to exit"
 script = """

ci/scripts/deterministic-recovery-test.sh

@@ -9,11 +9,19 @@ echo "--- Download artifacts"
 download-and-decompress-artifact risingwave_simulation .
 chmod +x ./risingwave_simulation
 
-export RUST_LOG="info,risingwave_meta::barrier::recovery=debug"
+export RUST_LOG="info,\
+risingwave_meta::barrier::recovery=debug,\
+risingwave_meta::rpc::ddl_controller=debug,\
+risingwave_meta::barrier::mod=debug,\
+risingwave_simulation=debug"
 export LOGDIR=.risingwave/log
 
 mkdir -p $LOGDIR
 
+# FIXME(kwannoel): Why is this failing?
+# echo "--- deterministic simulation e2e, ci-3cn-2fe-3meta, recovery, background_ddl"
+# seq $TEST_NUM | parallel MADSIM_TEST_SEED={} './risingwave_simulation --kill --kill-rate=${KILL_RATE} ./e2e_test/background_ddl/sim/basic.slt 2> $LOGDIR/recovery-ddl-{}.log && rm $LOGDIR/recovery-ddl-{}.log'
+
 echo "--- deterministic simulation e2e, ci-3cn-2fe-3meta, recovery, ddl"
 seq $TEST_NUM | parallel MADSIM_TEST_SEED={} './risingwave_simulation --kill --kill-rate=${KILL_RATE} ./e2e_test/ddl/\*\*/\*.slt 2> $LOGDIR/recovery-ddl-{}.log && rm $LOGDIR/recovery-ddl-{}.log'
 

ci/scripts/e2e-kafka-sink-test.sh

@@ -5,6 +5,7 @@ set -euo pipefail
 
 ./.risingwave/bin/kafka/bin/kafka-topics.sh --bootstrap-server 127.0.0.1:29092 --topic test-rw-sink-append-only --create > /dev/null 2>&1
 ./.risingwave/bin/kafka/bin/kafka-topics.sh --bootstrap-server 127.0.0.1:29092 --topic test-rw-sink-upsert --create > /dev/null 2>&1
+./.risingwave/bin/kafka/bin/kafka-topics.sh --bootstrap-server 127.0.0.1:29092 --topic test-rw-sink-upsert-schema --create > /dev/null 2>&1
 ./.risingwave/bin/kafka/bin/kafka-topics.sh --bootstrap-server 127.0.0.1:29092 --topic test-rw-sink-debezium --create > /dev/null 2>&1
 
 sqllogictest -p 4566 -d dev 'e2e_test/sink/kafka/create_sink.slt'
@@ -28,6 +29,15 @@ if [ $? -ne 0 ]; then
   exit 1
 fi
 
+# test upsert kafka sink with schema
+echo "testing upsert kafka sink with schema"
+diff ./e2e_test/sink/kafka/upsert_schema1.result \
+<((./.risingwave/bin/kafka/bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:29092 --topic test-rw-sink-upsert-schema --from-beginning --property print.key=true --max-messages 10 | sort) 2> /dev/null)
+if [ $? -ne 0 ]; then
+  echo "The output for upsert sink with schema is not as expected."
+  exit 1
+fi
+
 # test debezium kafka sink
 echo "testing debezium kafka sink"
 (./.risingwave/bin/kafka/bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:29092 --topic test-rw-sink-debezium --property print.key=true --from-beginning --max-messages 10 | sort) > ./e2e_test/sink/kafka/debezium1.tmp.result 2> /dev/null
@@ -62,6 +72,15 @@ if [ $? -ne 0 ]; then
   exit 1
 fi
 
+# test upsert kafka sink with schema after update
+echo "testing upsert kafka sink with schema after updating data"
+diff ./e2e_test/sink/kafka/upsert_schema2.result \
+<((./.risingwave/bin/kafka/bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:29092 --topic test-rw-sink-upsert-schema --from-beginning --property print.key=true --max-messages 11 | sort) 2> /dev/null)
+if [ $? -ne 0 ]; then
+  echo "The output for upsert sink with schema is not as expected."
+  exit 1
+fi
+
 # test debezium kafka sink after update
 echo "testing debezium kafka sink after updating data"
 (./.risingwave/bin/kafka/bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:29092 --topic test-rw-sink-debezium --property print.key=true --from-beginning --max-messages 11  | sort) > ./e2e_test/sink/kafka/debezium2.tmp.result 2> /dev/null
@@ -87,6 +106,15 @@ if [ $? -ne 0 ]; then
   exit 1
 fi
 
+# test upsert kafka sink with schema after delete
+echo "testing upsert kafka sink with schema after deleting data"
+diff ./e2e_test/sink/kafka/upsert_schema3.result \
+<((./.risingwave/bin/kafka/bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:29092 --topic test-rw-sink-upsert-schema --from-beginning --property print.key=true --max-messages 12 | sort) 2> /dev/null)
+if [ $? -ne 0 ]; then
+  echo "The output for upsert sink with schema is not as expected."
+  exit 1
+fi
+
 # test debezium kafka sink after delete
 echo "testing debezium kafka sink after deleting data"
 (./.risingwave/bin/kafka/bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:29092 --topic test-rw-sink-debezium --property print.key=true --from-beginning --max-messages 13 | sort) > ./e2e_test/sink/kafka/debezium3.tmp.result 2> /dev/null
@@ -103,3 +131,10 @@ sqllogictest -p 4566 -d dev 'e2e_test/sink/kafka/drop_sink.slt'
 ./.risingwave/bin/kafka/bin/kafka-topics.sh --bootstrap-server 127.0.0.1:29092 --topic test-rw-sink-append-only --delete > /dev/null 2>&1
 ./.risingwave/bin/kafka/bin/kafka-topics.sh --bootstrap-server 127.0.0.1:29092 --topic test-rw-sink-upsert --delete > /dev/null 2>&1
 ./.risingwave/bin/kafka/bin/kafka-topics.sh --bootstrap-server 127.0.0.1:29092 --topic test-rw-sink-debezium --delete > /dev/null 2>&1
+
+# test different encoding
+echo "testing protobuf"
+cp src/connector/src/test_data/proto_recursive/recursive.pb ./proto-recursive
+./.risingwave/bin/kafka/bin/kafka-topics.sh --bootstrap-server 127.0.0.1:29092 --topic test-rw-sink-append-only-protobuf --create > /dev/null 2>&1
+sqllogictest -p 4566 -d dev 'e2e_test/sink/kafka/protobuf.slt'
+./.risingwave/bin/kafka/bin/kafka-topics.sh --bootstrap-server 127.0.0.1:29092 --topic test-rw-sink-append-only-protobuf --delete > /dev/null 2>&1