fix

Signed-off-by: xxchan <[email protected]>

Author: xxchan

src/bench/sink_bench/main.rs

@@ -56,10 +56,10 @@ use risingwave_connector::sink::{
     SINK_TYPE_APPEND_ONLY, SINK_TYPE_UPSERT,
 };
 use risingwave_connector::source::datagen::{
-    DatagenProperties, DatagenSplitEnumerator, DatagenSplitReader,
+    DatagenProperties, DatagenSplit, DatagenSplitEnumerator, DatagenSplitReader,
 };
 use risingwave_connector::source::{
-    Column, DataType, SourceContext, SourceEnumeratorContext, SplitReader,
+    Column, DataType, SourceContext, SourceEnumeratorContext, SplitEnumerator, SplitReader,
 };
 use risingwave_stream::executor::test_utils::prelude::ColumnDesc;
 use risingwave_stream::executor::{Barrier, Message, MessageStreamItem, StreamExecutorError};
@@ -303,7 +303,7 @@ impl MockDatagenSource {
             datagen_split_readers.push(
                 DatagenSplitReader::new(
                     properties.clone(),
-                    vec![splits],
+                    vec![DatagenSplit::try_from(splits).unwrap()],
                     parser_config.clone(),
                     SourceContext::dummy().into(),
                     Some(source_schema.clone()),

src/meta/service/src/cloud_service.rs

@@ -18,10 +18,8 @@ use std::sync::LazyLock;
 use async_trait::async_trait;
 use regex::Regex;
 use risingwave_connector::error::ConnectorResult;
-use risingwave_connector::source::{
-    ConnectorProperties, SourceEnumeratorContext, SourceProperties, SplitEnumerator,
-};
-use risingwave_connector::{dispatch_source_prop, WithOptionsSecResolved};
+use risingwave_connector::source::{ConnectorProperties, SourceEnumeratorContext, SplitEnumerator};
+use risingwave_connector::WithOptionsSecResolved;
 use risingwave_pb::cloud_service::cloud_service_server::CloudService;
 use risingwave_pb::cloud_service::rw_cloud_validate_source_response::{Error, ErrorType};
 use risingwave_pb::cloud_service::{

src/meta/src/stream/source_manager.rs

@@ -27,7 +27,7 @@ use risingwave_common::metrics::LabelGuardedIntGauge;
 use risingwave_connector::error::ConnectorResult;
 use risingwave_connector::source::{
     fill_adaptive_split, ConnectorProperties, SourceEnumeratorContext, SourceEnumeratorInfo,
-    SourceProperties, SplitEnumerator, SplitId, SplitImpl, SplitMetaData,
+    SplitEnumerator, SplitId, SplitImpl, SplitMetaData,
 };
 use risingwave_connector::WithOptionsSecResolved;
 use risingwave_meta_model::SourceId;