chore(connector): add some log and fix typo (#2639)

* chore(connector): add some log and fix typo Signed-off-by: Peng Chen <[email protected]> * fix log Signed-off-by: Peng Chen <[email protected]>
risingwavelabs · May 18, 2022 · 767e616 · 767e616
1 parent be878ce
commit 767e616
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/src/source/src/connector_source.rs b/src/source/src/connector_source.rs
@@ -103,17 +103,20 @@ impl InnerConnectorSourceReader {
                 biased;
                 // stop chan has high priority
                 _ = stop.borrow_mut() => {
+                    log::debug!("connector reader {} stop signal received", self.split.id());
                     break;
                 }
 
                 chunk = self.reader.next() => {
                     match chunk.map_err(|e| internal_error(e.to_string())) {
                         Err(e) => {
+                            log::error!("connector reader {} error happened {}", self.split.id(), e.to_string());
                             // just ignore
                             let _ = output.send(Either::Right(e));
                             break;
                         },
                         Ok(None) => {
+                            log::warn!("connector reader {} stream stopped", self.split.id());
                             break;
                         },
                         Ok(Some(msg)) => {
@@ -254,7 +257,7 @@ impl ConnectorSource {
         let columns = self.get_target_columns(column_ids)?;
 
         let readers = try_join_all(splits.into_iter().map(|split| {
-            log::debug!("spawning pulsar split reader for split {:?}", split);
+            log::debug!("spawning connector split reader for split {:?}", split);
             let props = config.clone();
             let columns = columns.clone();
             async move { InnerConnectorSourceReader::new(props, split, columns).await }