fix(test): add timestamp to kafka messages in the simulation (#7778)

This PR includes the `kafka_batch.slt` simulation test in CI, and adds timestamp to kafka messages to make it pass.

Approved-By: BugenZhao

Author: wangrunji0408

ci/scripts/deterministic-e2e-test.sh

@@ -30,7 +30,7 @@ echo "--- deterministic simulation e2e, ci-3cn-2fe, batch"
 seq $TEST_NUM | parallel MADSIM_TEST_SEED={} './risingwave_simulation ./e2e_test/batch/\*\*/\*.slt 2> $LOGDIR/batch-{}.log && rm $LOGDIR/batch-{}.log'
 
 echo "--- deterministic simulation e2e, ci-3cn-2fe, kafka source"
-seq $TEST_NUM | parallel MADSIM_TEST_SEED={} './risingwave_simulation --kafka-datadir=./scripts/source/test_data ./e2e_test/source/basic/kafka.slt 2> $LOGDIR/source-{}.log && rm $LOGDIR/source-{}.log'
+seq $TEST_NUM | parallel MADSIM_TEST_SEED={} './risingwave_simulation --kafka-datadir=./scripts/source/test_data ./e2e_test/source/basic/kafka\*.slt 2> $LOGDIR/source-{}.log && rm $LOGDIR/source-{}.log'
 
 echo "--- deterministic simulation e2e, ci-3cn-2fe, parallel, streaming"
 seq $TEST_NUM | parallel MADSIM_TEST_SEED={} './risingwave_simulation -j 16 ./e2e_test/streaming/\*\*/\*.slt 2> $LOGDIR/parallel-streaming-{}.log && rm $LOGDIR/parallel-streaming-{}.log'

src/tests/simulation/src/kafka.rs

@@ -13,9 +13,9 @@
 // limitations under the License.
 
 use std::collections::HashMap;
+use std::time::SystemTime;
 
 use rdkafka::admin::{AdminClient, AdminOptions, NewTopic, TopicReplication};
-use rdkafka::consumer::StreamConsumer;
 use rdkafka::error::{KafkaError, RDKafkaErrorCode};
 use rdkafka::producer::{BaseProducer, BaseRecord};
 use rdkafka::ClientConfig;
@@ -78,10 +78,19 @@ pub async fn producer(broker_addr: &str, datadir: String) {
             .expect("failed to create topic");
 
         let content = std::fs::read(file.path()).unwrap();
-        // binary message data, a file is a message
-        if topic.ends_with("bin") {
+        let msgs: Box<dyn Iterator<Item = &[u8]> + Send> = if topic.ends_with("bin") {
+            // binary message data, a file is a message
+            Box::new(std::iter::once(content.as_slice()))
+        } else {
+            Box::new(content.split(|&b| b == b'\n'))
+        };
+        for msg in msgs {
             loop {
-                let record = BaseRecord::<(), _>::to(topic).payload(&content);
+                let ts = SystemTime::now()
+                    .duration_since(SystemTime::UNIX_EPOCH)
+                    .unwrap()
+                    .as_millis() as i64;
+                let record = BaseRecord::<(), _>::to(topic).payload(msg).timestamp(ts);
                 match producer.send(record) {
                     Ok(_) => break,
                     Err((KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull), _)) => {
@@ -90,19 +99,6 @@ pub async fn producer(broker_addr: &str, datadir: String) {
                     Err((e, _)) => panic!("failed to send message: {}", e),
                 }
             }
-        } else {
-            for line in content.split(|&b| b == b'\n') {
-                loop {
-                    let record = BaseRecord::<(), _>::to(topic).payload(line);
-                    match producer.send(record) {
-                        Ok(_) => break,
-                        Err((KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull), _)) => {
-                            producer.flush(None).await.expect("failed to flush");
-                        }
-                        Err((e, _)) => panic!("failed to send message: {}", e),
-                    }
-                }
-            }
         }
         producer.flush(None).await.expect("failed to flush");
     }