fix(batch, source): Propagate user errors (risingwavelabs#8493)

Signed-off-by: Runji Wang <[email protected]>
Co-authored-by: jon-chuang <[email protected]>
Co-authored-by: Runji Wang <[email protected]>

Author: jon-chuang

src/connector/src/macros.rs

@@ -182,8 +182,9 @@ macro_rules! impl_common_parser_logic {
                             if let Err(e) = self.parse_inner(content.as_ref(), builder.row_writer())
                                 .await
                             {
-                                self.source_ctx.report_stream_source_error(&e);
                                 tracing::warn!("message parsing failed {}, skipping", e.to_string());
+                                // This will throw an error for batch
+                                self.source_ctx.report_user_source_error(e)?;
                                 continue;
                             }
 

src/connector/src/source/base.rs

@@ -25,7 +25,7 @@ use parking_lot::Mutex;
 use prost::Message;
 use risingwave_common::array::StreamChunk;
 use risingwave_common::catalog::TableId;
-use risingwave_common::error::{ErrorCode, ErrorSuppressor, RwError};
+use risingwave_common::error::{ErrorCode, ErrorSuppressor, Result as RwResult, RwError};
 use risingwave_pb::connector_service::TableSchema;
 use risingwave_pb::source::ConnectorSplit;
 use serde::{Deserialize, Serialize};
@@ -107,10 +107,10 @@ impl SourceContext {
         self.error_suppressor = Some(error_suppressor)
     }
 
-    pub(crate) fn report_stream_source_error(&self, e: &RwError) {
-        // Do not report for batch
+    pub(crate) fn report_user_source_error(&self, e: RwError) -> RwResult<()> {
+        // Repropagate the error if batch
         if self.source_info.fragment_id == u32::MAX {
-            return;
+            return Err(e);
         }
         let mut err_str = e.inner().to_string();
         if let Some(suppressor) = &self.error_suppressor &&
@@ -130,6 +130,7 @@ impl SourceContext {
                 &self.source_info.source_id.table_id.to_string(),
             ])
             .inc();
+        Ok(())
     }
 }
 

src/tests/simulation/src/kafka.rs

@@ -82,7 +82,11 @@ pub async fn producer(broker_addr: &str, datadir: String) {
             // 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'))
+            Box::new(
+                content
+                    .split(|&b| b == b'\n')
+                    .filter(|line| !line.is_empty()),
+            )
         };
         for msg in msgs {
             loop {