fix: use error instead of warn for source/sink errors (#23178) (#23189)

Co-authored-by: Zhanxiang (Patrick) Huang <hzxa21@hotmail.com>
Co-authored-by: Chengyou Liu <35356271+cyliu0@users.noreply.github.com>

Author: 3 people

src/stream/src/executor/sink.rs

@@ -660,7 +660,7 @@ impl<F: LogStoreFactory> SinkExecutor<F> {
                     if F::ALLOW_REWIND {
                         match log_reader.rewind().await {
                             Ok(()) => {
-                                warn!(
+                                error!(
                                     error = %e.as_report(),
                                     executor_id = sink_writer_param.executor_id,
                                     sink_id = sink_param.sink_id.sink_id,

src/stream/src/executor/source/reader_stream.rs

@@ -17,6 +17,7 @@ use std::time::Duration;
 
 use itertools::Itertools;
 use risingwave_common::catalog::{ColumnId, TableId};
+use risingwave_common::metrics::GLOBAL_ERROR_METRICS;
 use risingwave_connector::parser::schema_change::SchemaChangeEnvelope;
 use risingwave_connector::source::reader::desc::SourceDesc;
 use risingwave_connector::source::{
@@ -179,13 +180,19 @@ impl StreamReaderBuilder {
                 if is_initial_build {
                     return Err(StreamExecutorError::connector_error(e));
                 } else {
-                    tracing::warn!(
+                    tracing::error!(
                         error = %e.as_report(),
                         source_name = self.source_name,
                         source_id = self.source_id.table_id,
                         actor_id = self.actor_ctx.id,
                         "build stream source reader error, retry in 1s"
                     );
+                    GLOBAL_ERROR_METRICS.user_source_error.report([
+                        e.variant_name().to_owned(),
+                        self.source_id.table_id.to_string(),
+                        self.source_name.to_owned(),
+                        self.actor_ctx.fragment_id.to_string(),
+                    ]);
                     tokio::time::sleep(Duration::from_secs(1)).await;
                     continue 'build_consume_loop;
                 }
@@ -208,13 +215,19 @@ impl StreamReaderBuilder {
                         yield (msg, latest_splits_info.clone());
                     }
                     Err(e) => {
-                        tracing::warn!(
+                        tracing::error!(
                             error = %e.as_report(),
                             source_name = self.source_name,
                             source_id = self.source_id.table_id,
                             actor_id = self.actor_ctx.id,
                             "stream source reader error"
                         );
+                        GLOBAL_ERROR_METRICS.user_source_error.report([
+                            e.variant_name().to_owned(),
+                            self.source_id.table_id.to_string(),
+                            self.source_name.to_owned(),
+                            self.actor_ctx.fragment_id.to_string(),
+                        ]);
                         is_error = true;
                         break 'consume;
                     }

src/stream/src/executor/source/source_backfill_executor.rs

@@ -497,7 +497,7 @@ impl<S: StateStore> SourceBackfillExecutorInner<S> {
                     Either::Left(msg) => {
                         let Ok(msg) = msg else {
                             let e = msg.unwrap_err();
-                            tracing::warn!(
+                            tracing::error!(
                                 error = ?e.as_report(),
                                 source_id = %self.source_id,
                                 "stream source reader error",

src/stream/src/executor/source/source_executor.rs

@@ -376,7 +376,7 @@ impl<S: StateStore> SourceExecutor<S> {
         e: StreamExecutorError,
     ) -> StreamExecutorResult<()> {
         let core = &mut self.stream_source_core;
-        tracing::warn!(
+        tracing::error!(
             error = ?e.as_report(),
             actor_id = self.actor_ctx.id,
             source_id = %core.source_id,