report error with score

risingwavelabs · Dec 4, 2024 · d28b483 · d28b483
1 parent ae53d6d
commit d28b483
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 8 deletions.
diff --git a/src/stream/src/task/barrier_manager.rs b/src/stream/src/task/barrier_manager.rs
@@ -63,7 +63,7 @@ use risingwave_pb::stream_service::streaming_control_stream_response::{
 };
 use risingwave_pb::stream_service::{
     streaming_control_stream_response, BarrierCompleteResponse, InjectBarrierRequest,
-    StreamingControlStreamRequest, StreamingControlStreamResponse,
+    PbScoredError, StreamingControlStreamRequest, StreamingControlStreamResponse,
 };
 
 use crate::executor::exchange::permit::Receiver;
@@ -160,12 +160,15 @@ impl ControlStreamHandle {
     pub(super) fn ack_reset_database(
         &mut self,
         database_id: DatabaseId,
-        root_err: Option<StreamError>,
+        root_err: Option<ScoredStreamError>,
         reset_request_id: u32,
     ) {
         self.send_response(Response::ResetDatabase(ResetDatabaseResponse {
             database_id: database_id.database_id,
-            root_err: root_err.map(|err| err.to_report_string()),
+            root_err: root_err.map(|err| PbScoredError {
+                err_msg: err.error.to_report_string(),
+                score: err.score.0,
+            }),
             reset_request_id,
         }));
     }

diff --git a/src/stream/src/task/barrier_manager/managed_state.rs b/src/stream/src/task/barrier_manager/managed_state.rs
@@ -82,7 +82,7 @@ use risingwave_pb::stream_service::streaming_control_stream_request::{
 use risingwave_pb::stream_service::InjectBarrierRequest;
 
 use crate::task::barrier_manager::await_epoch_completed_future::AwaitEpochCompletedFuture;
-use crate::task::barrier_manager::LocalBarrierEvent;
+use crate::task::barrier_manager::{LocalBarrierEvent, ScoredStreamError};
 
 pub(super) struct ManagedBarrierStateDebugInfo<'a> {
     running_actors: BTreeSet<ActorId>,
@@ -361,9 +361,7 @@ impl SuspendedDatabaseState {
         if let Some(hummock) = self.inner.actor_manager.env.state_store().as_hummock() {
             hummock.clear_tables(self.inner.table_ids).await;
         }
-        ResetDatabaseOutput {
-            root_err: root_err.error,
-        }
+        ResetDatabaseOutput { root_err }
     }
 }
 
@@ -373,7 +371,7 @@ pub(crate) struct ResettingDatabaseState {
 }
 
 pub(crate) struct ResetDatabaseOutput {
-    pub(crate) root_err: StreamError,
+    pub(crate) root_err: ScoredStreamError,
 }
 
 pub(crate) enum DatabaseStatus {