fix(source): parse message with empty key and payload

ci/scripts/e2e-source-test.sh

@@ -72,10 +72,20 @@ echo "--- mysql & postgres cdc validate test"
 sqllogictest -p 4566 -d dev './e2e_test/source/cdc/cdc.validate.mysql.slt'
 sqllogictest -p 4566 -d dev './e2e_test/source/cdc/cdc.validate.postgres.slt'
 
+echo "--- cdc share source test"
 # cdc share stream test cases
 export MYSQL_HOST=mysql MYSQL_TCP_PORT=3306 MYSQL_PWD=123456
 sqllogictest -p 4566 -d dev './e2e_test/source/cdc/cdc.share_stream.slt'
 
+# create a share source and check whether heartbeat message is received
+sqllogictest -p 4566 -d dev './e2e_test/source/cdc/cdc.create_source_job.slt'
+table_id=`psql -U root -h localhost -p 4566 -d dev -t -c "select id from rw_internal_tables where name like '%mysql_source%';" | xargs`;
+table_count=`psql -U root -h localhost -p 4566 -d dev -t -c "select count(*) from rw_table(${table_id}, public);" | xargs`;
+if [ $table_count -eq 0 ]; then
+    echo "ERROR: internal table of cdc share source is empty!"
+    exit 1
+fi
+
 echo "--- mysql & postgres load and check"
 sqllogictest -p 4566 -d dev './e2e_test/source/cdc/cdc.load.slt'
 # wait for cdc loading

e2e_test/source/cdc/cdc.create_source_job.slt

@@ -0,0 +1,15 @@
+control substitution on
+
+# create a cdc source job, which format fixed to `FORMAT PLAIN ENCODE JSON`
+statement ok
+create source mysql_source with (
+ connector = 'mysql-cdc',
+ hostname = '${MYSQL_HOST:localhost}',
+ port = '${MYSQL_TCP_PORT:8306}',
+ username = 'rwcdc',
+ password = '${MYSQL_PWD:}',
+ database.name = 'mytest',
+ server.id = '5001'
+);
+
+sleep 2s

java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/source/SourceValidateHandler.java

@@ -95,6 +95,11 @@ public static void validateSource(ConnectorServiceProto.ValidateSourceRequest re
 
         boolean isCdcSourceJob = request.getIsSourceJob();
         boolean isBackfillTable = request.getIsBackfillTable();
+        LOG.info(
+                "source_id: {}, isCdcSourceJob: {}, isBackfillTable: {}",
+                request.getSourceId(),
+                isCdcSourceJob,
+                isBackfillTable);
 
         TableSchema tableSchema = TableSchema.fromProto(request.getTableSchema());
         switch (request.getSourceType()) {

src/connector/src/parser/mod.rs

@@ -80,6 +80,7 @@ mod upsert_parser;
 mod util;
 
 pub use debezium::DEBEZIUM_IGNORE_KEY;
+use risingwave_common::buffer::BitmapBuilder;
 pub use unified::{AccessError, AccessResult};
 
 /// A builder for building a [`StreamChunk`] from [`SourceColumnDesc`].
@@ -132,6 +133,25 @@ impl SourceStreamChunkBuilder {
         builder.finish()
     }
 
+    fn finish_heartbeat(self) -> StreamChunk {
+        // heartbeat chunk should be invisible
+        let visibility = BitmapBuilder::zeroed(self.op_builder.len());
+        StreamChunk::with_visibility(
+            self.op_builder,
+            self.builders
+                .into_iter()
+                .map(|builder| builder.finish().into())
+                .collect(),
+            visibility.finish(),
+        )
+    }
+
+    fn take_heartbeat(&mut self, next_cap: usize) -> StreamChunk {
+        let descs = std::mem::take(&mut self.descs);
+        let builder = std::mem::replace(self, Self::with_capacity(descs, next_cap));
+        builder.finish_heartbeat()
+    }
+
     pub fn op_num(&self) -> usize {
         self.op_builder.len()
     }
@@ -560,6 +580,10 @@ pub trait ByteStreamSourceParser: Send + Debug + Sized + 'static {
         self.parse_one(key, payload, writer)
             .map_ok(|_| ParseResult::Rows)
     }
+
+    fn emit_empty_row<'a>(&'a mut self, mut writer: SourceStreamChunkRowWriter<'a>) {
+        _ = writer.insert(|_column| Ok(None));
+    }
 }
 
 #[try_stream(ok = Vec<SourceMessage>, error = ConnectorError)]
@@ -616,6 +640,7 @@ const MAX_ROWS_FOR_TRANSACTION: usize = 4096;
 async fn into_chunk_stream<P: ByteStreamSourceParser>(mut parser: P, data_stream: BoxSourceStream) {
     let columns = parser.columns().to_vec();
 
+    let mut heartbeat_builder = SourceStreamChunkBuilder::with_capacity(columns.clone(), 0);
     let mut builder = SourceStreamChunkBuilder::with_capacity(columns, 0);
 
     struct Transaction {
@@ -657,7 +682,15 @@ async fn into_chunk_stream<P: ByteStreamSourceParser>(mut parser: P, data_stream
         let process_time_ms = chrono::Utc::now().timestamp_millis();
         for (i, msg) in batch.into_iter().enumerate() {
             if msg.key.is_none() && msg.payload.is_none() {
-                tracing::debug!(offset = msg.offset, "skip parsing of heartbeat message");
+                tracing::debug!(
+                    offset = msg.offset,
+                    "got a empty message, could be a heartbeat"
+                );
+                parser.emit_empty_row(heartbeat_builder.row_writer().with_meta(MessageMeta {
+                    meta: &msg.meta,
+                    split_id: &msg.split_id,
+                    offset: &msg.offset,
+                }));
                 continue;
             }
 
@@ -750,6 +783,13 @@ async fn into_chunk_stream<P: ByteStreamSourceParser>(mut parser: P, data_stream
             }
         }
 
+        // emit heartbeat for each message batch
+        // we must emit heartbeat chunk before the data chunk,
+        // otherwise the source offset could be backward due to the heartbeat
+        if !heartbeat_builder.is_empty() {
+            yield heartbeat_builder.take_heartbeat(0);
+        }
+
         // If we are not in a transaction, we should yield the chunk now.
         if current_transaction.is_none() {
             yield_asap = false;

src/stream/src/executor/source/mod.rs

@@ -60,10 +60,22 @@ pub fn get_split_offset_mapping_from_chunk(
     offset_idx: usize,
 ) -> Option<HashMap<SplitId, String>> {
     let mut split_offset_mapping = HashMap::new();
-    for (_, row) in chunk.rows() {
-        let split_id = row.datum_at(split_idx).unwrap().into_utf8().into();
-        let offset = row.datum_at(offset_idx).unwrap().into_utf8();
-        split_offset_mapping.insert(split_id, offset.to_string());
+    if chunk.cardinality() == 0 {
+        // assumes the chunk is a heartbeat chunk
+        for i in 0..chunk.capacity() {
+            let (_, row, _) = chunk.row_at(i);
+            let split_id = row.datum_at(split_idx).unwrap().into_utf8().into();
+            let offset = row.datum_at(offset_idx).unwrap().into_utf8();
+            split_offset_mapping
+                .entry(split_id)
+                .or_insert(offset.to_string());
+        }
+    } else {
+        for (_, row) in chunk.rows() {
+            let split_id = row.datum_at(split_idx).unwrap().into_utf8().into();
+            let offset = row.datum_at(offset_idx).unwrap().into_utf8();
+            split_offset_mapping.insert(split_id, offset.to_string());
+        }
     }
     Some(split_offset_mapping)
 }