test: move more Kafka tests to source_inline

e2e_test/source/basic/handling_mode.slt → e2e_test/source_inline/kafka/handling_mode.slt

@@ -1,9 +1,24 @@
+control substitution on
+
+system ok
+rpk topic delete json_timestamptz_handling_mode || true
+
+system ok
+rpk topic create json_timestamptz_handling_mode -p 1
+
+system ok
+cat <<EOF | rpk topic produce json_timestamptz_handling_mode -f "%v\n"
+{"case":"0 number small","payloade2e_test/source_inline/kafka/handling_mode.slt":{"after":{"case":"0 number small","at":100},"op":"r"}}
+{"case":"1 number recent","payload":{"after":{"case":"1 number recent","at":1712800800123456},"op":"r"}}
+{"case":"2 string utc","payload":{"after":{"case":"2 string utc","at":"2024-04-11T02:00:00.654321Z"},"op":"r"}}
+{"case":"3 string naive","payload":{"after":{"case":"3 string naive","at":"2024-04-11 02:00:00.234321"},"op":"r"}}
+EOF
+
 statement error unrecognized
 create table t (
   payload struct<after struct<"case" varchar, at timestamptz>>)
 with (
-  connector = 'kafka',
-  properties.bootstrap.server='message_queue:29092',
+  ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
   topic = 'json_timestamptz_handling_mode')
 format plain encode json (timestamptz.handling.mode = 'mili');
 
@@ -13,8 +28,7 @@ create table plain_guess (
   "case" varchar,
   payload struct<after struct<"case" varchar, at timestamptz>>)
 with (
-  connector = 'kafka',
-  properties.bootstrap.server='message_queue:29092',
+  ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
   topic = 'json_timestamptz_handling_mode')
 format plain encode json (timestamptz.handling.mod = 'mili');
 
@@ -23,8 +37,7 @@ create table plain_milli (
   "case" varchar,
   payload struct<after struct<"case" varchar, at timestamptz>>)
 with (
-  connector = 'kafka',
-  properties.bootstrap.server='message_queue:29092',
+  ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
   topic = 'json_timestamptz_handling_mode')
 format plain encode json (timestamptz.handling.mode = 'milli');
 
@@ -33,8 +46,7 @@ create table plain_micro (
   "case" varchar,
   payload struct<after struct<"case" varchar, at timestamptz>>)
 with (
-  connector = 'kafka',
-  properties.bootstrap.server='message_queue:29092',
+  ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
   topic = 'json_timestamptz_handling_mode')
 format plain encode json (timestamptz.handling.mode = 'micro');
 
@@ -43,8 +55,7 @@ create table plain_utc (
   "case" varchar,
   payload struct<after struct<"case" varchar, at timestamptz>>)
 with (
-  connector = 'kafka',
-  properties.bootstrap.server='message_queue:29092',
+  ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
   topic = 'json_timestamptz_handling_mode')
 format plain encode json (timestamptz.handling.mode = 'utc_string');
 
@@ -53,17 +64,15 @@ create table plain_naive (
   "case" varchar,
   payload struct<after struct<"case" varchar, at timestamptz>>)
 with (
-  connector = 'kafka',
-  properties.bootstrap.server='message_queue:29092',
+  ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
   topic = 'json_timestamptz_handling_mode')
 format plain encode json (timestamptz.handling.mode = 'utc_without_suffix');
 
 statement ok
 create table debezium_milli (
   "case" varchar, at timestamptz, primary key("case"))
 with (
-  connector = 'kafka',
-  properties.bootstrap.server='message_queue:29092',
+  ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
   topic = 'json_timestamptz_handling_mode')
 format debezium encode json (timestamptz.handling.mode = 'milli');
 

e2e_test/source/basic/temporary_kafka_batch.slt → e2e_test/source_inline/kafka/temporary_kafka_batch.slt

@@ -1,8 +1,23 @@
+control substitution on
+
+system ok
+rpk topic delete test_temporary_kafka_batch || true
+
+system ok
+rpk topic create test_temporary_kafka_batch -p 1
+
+system ok
+cat <<EOF | rpk topic produce test_temporary_kafka_batch -f "%v\n"
+{"v1": 1, "v2": "1"}
+{"v1": 2, "v2": "22"}
+{"v1": 3, "v2": "333"}
+{"v1": 4, "v2": "4444"}
+EOF
+
 statement ok
 create temporary source s1 (v1 int, v2 varchar) with (
-  connector = 'kafka',
-  topic = 'kafka_1_partition_topic',
-  properties.bootstrap.server = 'message_queue:29092',
+  ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
+  topic = 'test_temporary_kafka_batch',
   scan.startup.mode = 'earliest'
 ) FORMAT PLAIN ENCODE JSON
 

e2e_test/source_inline/kafka/ttl_table_with_con.slt

@@ -0,0 +1,47 @@
+control substitution on
+
+system ok
+rpk topic delete test_ttl_table_with_con || true
+
+system ok
+rpk topic create test_ttl_table_with_con -p 1
+
+system ok
+cat <<EOF | rpk topic produce test_ttl_table_with_con -f "%v\n"
+{"v1": 1, "v2": "1"}
+{"v1": 2, "v2": "22"}
+{"v1": 3, "v2": "333"}
+{"v1": 4, "v2": "4444"}
+EOF
+
+statement ok
+create table t (v1 int, v2 varchar) APPEND ONLY with (
+  ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
+  topic = 'test_ttl_table_with_con',
+  scan.startup.mode = 'earliest',
+  retention_seconds = 5
+) FORMAT PLAIN ENCODE JSON;
+
+statement ok
+flush;
+
+# Wait enough time to ensure SourceExecutor consumes all Kafka data.
+sleep 1s
+
+query IT rowsort
+select * from t
+----
+1 1
+2 22
+3 333
+4 4444
+
+statement ok
+select pg_sleep(10);
+
+query I
+select * from t;
+----
+
+statement ok
+drop table t;