refactor(meta): splits source_manager into smaller mods (#20071)

Signed-off-by: xxchan <xxchan22f@gmail.com>

Author: xxchan

src/meta/src/rpc/ddl_controller.rs

@@ -75,7 +75,7 @@ use crate::manager::{
 };
 use crate::model::{StreamContext, StreamJobFragments, TableParallelism};
 use crate::stream::{
-    create_source_worker_handle, validate_sink, ActorGraphBuildResult, ActorGraphBuilder,
+    create_source_worker, validate_sink, ActorGraphBuildResult, ActorGraphBuilder,
     CompleteStreamFragmentGraph, CreateStreamingJobContext, CreateStreamingJobOption,
     GlobalStreamManagerRef, ReplaceStreamJobContext, SourceChange, SourceManagerRef,
     StreamFragmentGraph,
@@ -441,7 +441,7 @@ impl DdlController {
 
     /// Shared source is handled in [`Self::create_streaming_job`]
     async fn create_non_shared_source(&self, source: Source) -> MetaResult<NotificationVersion> {
-        let handle = create_source_worker_handle(&source, self.source_manager.metrics.clone())
+        let handle = create_source_worker(&source, self.source_manager.metrics.clone())
             .await
             .context("failed to create source worker")?;