fix: propagate connector reader error to the upper layer (risingwavel…

…abs#8920) Signed-off-by: tabVersion <[email protected]>
Little-Wallace · Mar 31, 2023 · e51b240 · e51b240
1 parent 74b935e
commit e51b240
Show file tree

Hide file tree

Showing 2 changed files with 1 addition and 6 deletions.
diff --git a/src/source/src/connector_source.rs b/src/source/src/connector_source.rs
@@ -32,7 +32,6 @@ use risingwave_connector::source::{
 pub struct ConnectorSource {
     pub config: ConnectorProperties,
     pub columns: Vec<SourceColumnDesc>,
-    // pub parser: Arc<SourceParserImpl>,
     pub parser_config: SpecificParserConfig,
     pub connector_message_buffer_size: usize,
 }

diff --git a/src/stream/src/executor/stream_reader.rs b/src/stream/src/executor/stream_reader.rs
@@ -15,7 +15,6 @@
 use std::pin::Pin;
 use std::task::Poll;
 
-use await_tree::InstrumentAwait;
 use either::Either;
 use futures::stream::{select_with_strategy, BoxStream, PollNext, SelectWithStrategy};
 use futures::{Stream, StreamExt, TryStreamExt};
@@ -58,10 +57,7 @@ impl<const BIASED: bool> StreamReaderWithPause<BIASED> {
             match chunk {
                 Ok(chunk) => yield chunk,
                 Err(err) => {
-                    error!("hang up stream reader due to polling error: {}", err);
-                    futures::future::pending()
-                        .instrument_await("source_error")
-                        .await
+                    return Err(StreamExecutorError::connector_error(err));
                 }
             }
         }