fix

Signed-off-by: xxchan <[email protected]>

e2e_test/source_inline/kafka/protobuf/basic.slt

@@ -1,21 +1,21 @@
 control substitution on
 
 system ok
-rpk topic delete sr_pb_test || true; \
-(rpk sr subject delete 'sr_pb_test-value' && rpk sr subject delete 'sr_pb_test-value' --permanent) || true;
+rpk topic delete pb_alter_source_test || true; \
+(rpk sr subject delete 'pb_alter_source_test-value' && rpk sr subject delete 'pb_alter_source_test-value' --permanent) || true;
 
 system ok
-python3 e2e_test/source_inline/kafka/protobuf/pb.py "${RISEDEV_KAFKA_BOOTSTRAP_SERVERS}" "${RISEDEV_SCHEMA_REGISTRY_URL}" "sr_pb_test" 20 user
+python3 e2e_test/source_inline/kafka/protobuf/pb.py "${RISEDEV_KAFKA_BOOTSTRAP_SERVERS}" "${RISEDEV_SCHEMA_REGISTRY_URL}" "pb_alter_source_test" 20 user
 
 # make sure google/protobuf/source_context.proto is NOT in schema registry
 system ok
 curl --silent "${RISEDEV_SCHEMA_REGISTRY_URL}/subjects" | grep -v 'google/protobuf/source_context.proto'
 
 # Create a table.
 statement ok
-create table sr_pb_test with (
+create table pb_alter_source_test with (
     ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
-    topic = 'sr_pb_test',
+    topic = 'pb_alter_source_test',
     scan.startup.mode = 'earliest')
 FORMAT plain ENCODE protobuf(
         schema.registry = '${RISEDEV_SCHEMA_REGISTRY_URL}',
@@ -24,9 +24,9 @@ FORMAT plain ENCODE protobuf(
 
 # for multiple schema registry nodes
 statement ok
-create table sr_pb_test_bk with (
+create table pb_alter_source_test_bk with (
     ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
-    topic = 'sr_pb_test',
+    topic = 'pb_alter_source_test',
     scan.startup.mode = 'earliest')
 FORMAT plain ENCODE protobuf(
         schema.registry = '${RISEDEV_SCHEMA_REGISTRY_URL},${RISEDEV_SCHEMA_REGISTRY_URL}',
@@ -42,7 +42,7 @@ include
     key as rw_key
 with (
     ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
-    topic = 'sr_pb_test',
+    topic = 'pb_alter_source_test',
     scan.startup.mode = 'earliest')
 FORMAT upsert ENCODE protobuf(
         schema.registry = '${RISEDEV_SCHEMA_REGISTRY_URL}',
@@ -57,12 +57,12 @@ statement ok
 flush;
 
 query I
-select count(*) from sr_pb_test;
+select count(*) from pb_alter_source_test;
 ----
 20
 
 query IT
-select min(id), max(id), max((sc).file_name) from sr_pb_test;
+select min(id), max(id), max((sc).file_name) from pb_alter_source_test;
 ----
 0 19 source/context_019.proto
 
@@ -72,10 +72,10 @@ select convert_from(min(rw_key), 'UTF-8'), convert_from(max(rw_key), 'UTF-8') fr
 {"id": 0} {"id": 9}
 
 statement ok
-drop table sr_pb_test;
+drop table pb_alter_source_test;
 
 statement ok
-drop table sr_pb_test_bk;
+drop table pb_alter_source_test_bk;
 
 statement ok
 drop table sr_pb_upsert;

e2e_test/source_inline/kafka/temporary_kafka_batch.slt

@@ -71,6 +71,11 @@ t
 t
 t
 
+sleep 1s
+
+statement ok
+flush;
+
 query B
 select _rw_kafka_timestamp < now() from s1
 ----