feat: auto rebuild source reader in executor (#11993)

Co-authored-by: Bohan Zhang <[email protected]>

Author: wugouzi

docker/dashboards/risingwave-dev-dashboard.json

@@ -1443,6 +1443,16 @@ def section_streaming_errors(outer_panels):
                         ),
                     ],
                 ),
+                panels.timeseries_count(
+                    "Source Reader Errors by Type",
+                    "",
+                    [
+                        panels.target(
+                            f"sum({metric('user_source_reader_error_count')}) by (error_type, error_msg, actor_id, source_id, executor_name)",
+                            "{{error_type}}: {{error_msg}} ({{executor_name}}: actor_id={{actor_id}}, source_id={{source_id}})",
+                        ),
+                    ],
+                ),
             ],
         ),
     ]

grafana/risingwave-dev-dashboard.dashboard.py

@@ -517,8 +517,8 @@ async fn into_chunk_stream<P: ByteStreamSourceParser>(mut parser: P, data_stream
 
                 Err(error) => {
                     tracing::warn!(%error, "message parsing failed, skipping");
-                    // This will throw an error for batch
-                    parser.source_ctx().report_user_source_error(error)?;
+                    // Skip for batch
+                    parser.source_ctx().report_user_source_error(error);
                     continue;
                 }
             }

grafana/risingwave-dev-dashboard.json

@@ -25,7 +25,7 @@ use itertools::Itertools;
 use parking_lot::Mutex;
 use risingwave_common::array::StreamChunk;
 use risingwave_common::catalog::TableId;
-use risingwave_common::error::{ErrorCode, ErrorSuppressor, Result as RwResult, RwError};
+use risingwave_common::error::{ErrorCode, ErrorSuppressor, RwError};
 use risingwave_common::types::{JsonbVal, Scalar};
 use risingwave_pb::connector_service::PbTableSchema;
 use risingwave_pb::source::ConnectorSplit;
@@ -172,10 +172,9 @@ impl SourceContext {
         ctx
     }
 
-    pub(crate) fn report_user_source_error(&self, e: RwError) -> RwResult<()> {
-        // Repropagate the error if batch
+    pub(crate) fn report_user_source_error(&self, e: RwError) {
         if self.source_info.fragment_id == u32::MAX {
-            return Err(e);
+            return;
         }
         let mut err_str = e.inner().to_string();
         if let Some(suppressor) = &self.error_suppressor
@@ -198,7 +197,6 @@ impl SourceContext {
                 &self.source_info.source_id.table_id.to_string(),
             ])
             .inc();
-        Ok(())
     }
 }
 

src/connector/src/parser/mod.rs

@@ -121,6 +121,9 @@ pub struct StreamingMetrics {
     /// User compute error reporting
     pub user_compute_error_count: GenericCounterVec<AtomicU64>,
 
+    /// User source reader error
+    pub user_source_reader_error_count: GenericCounterVec<AtomicU64>,
+
     // Materialize
     pub materialize_cache_hit_count: GenericCounterVec<AtomicU64>,
     pub materialize_cache_total_count: GenericCounterVec<AtomicU64>,
@@ -669,6 +672,20 @@ impl StreamingMetrics {
         )
         .unwrap();
 
+        let user_source_reader_error_count = register_int_counter_vec_with_registry!(
+            "user_source_reader_error_count",
+            "Source reader error count",
+            &[
+                "error_type",
+                "error_msg",
+                "executor_name",
+                "actor_id",
+                "source_id"
+            ],
+            registry,
+        )
+        .unwrap();
+
         let materialize_cache_hit_count = register_int_counter_vec_with_registry!(
             "stream_materialize_cache_hit_count",
             "Materialize executor cache hit count",
@@ -761,6 +778,7 @@ impl StreamingMetrics {
             jemalloc_allocated_bytes,
             jemalloc_active_bytes,
             user_compute_error_count,
+            user_source_reader_error_count,
             materialize_cache_hit_count,
             materialize_cache_total_count,
             stream_memory_usage,