sync topic to topic in core V2 (#1684)

Author: phiSgr

apps/framework-cli/src/cli/display.rs

@@ -323,6 +323,15 @@ pub fn show_changes(infra_plan: &InfraPlan) {
         .processes_changes
         .iter()
         .for_each(|change| match change {
+            ProcessChange::TopicToTopicSyncProcess(Change::Added(infra)) => {
+                infra_added(&infra.expanded_display());
+            }
+            ProcessChange::TopicToTopicSyncProcess(Change::Removed(infra)) => {
+                infra_removed(&infra.short_display());
+            }
+            ProcessChange::TopicToTopicSyncProcess(Change::Updated { before, after: _ }) => {
+                infra_updated(&before.expanded_display());
+            }
             ProcessChange::TopicToTableSyncProcess(Change::Added(infra)) => {
                 infra_added(&infra.expanded_display());
             }

apps/framework-cli/src/framework/core/infrastructure.rs

@@ -6,7 +6,7 @@ pub mod function_process;
 pub mod olap_process;
 pub mod table;
 pub mod topic;
-pub mod topic_to_table_sync_process;
+pub mod topic_sync_process;
 pub mod view;
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]

apps/framework-cli/src/framework/core/infrastructure/topic_to_table_sync_process.rs renamed to apps/framework-cli/src/framework/core/infrastructure/topic_sync_process.rs

@@ -1,11 +1,11 @@
 use serde::{Deserialize, Serialize};
 
-use crate::framework::core::infrastructure_map::PrimitiveSignature;
-
 use super::{
     table::{Column, Table},
     topic::Topic,
 };
+use crate::framework::core::infrastructure_map::{PrimitiveSignature, PrimitiveTypes};
+use crate::framework::streaming::model::StreamingFunction;
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct TopicToTableSyncProcess {
@@ -18,6 +18,14 @@ pub struct TopicToTableSyncProcess {
     pub source_primitive: PrimitiveSignature,
 }
 
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct TopicToTopicSyncProcess {
+    pub source_topic_id: String,
+    pub target_topic_id: String,
+
+    pub source_primitive: PrimitiveSignature,
+}
+
 impl TopicToTableSyncProcess {
     pub fn new(topic: &Topic, table: &Table) -> Self {
         if topic.version != table.version {
@@ -57,3 +65,36 @@ impl TopicToTableSyncProcess {
         )
     }
 }
+
+impl TopicToTopicSyncProcess {
+    pub fn from_migration_function(function: &StreamingFunction) -> Self {
+        let source_topic = Topic::from_data_model(&function.source_data_model);
+        let (source_for_func, _) = Topic::from_migration_function(function);
+        TopicToTopicSyncProcess {
+            source_topic_id: source_topic.id(),
+            target_topic_id: source_for_func.id(),
+            source_primitive: PrimitiveSignature {
+                name: function.id(),
+                primitive_type: PrimitiveTypes::Function,
+            },
+        }
+    }
+
+    pub fn id(&self) -> String {
+        self.target_topic_id.to_string()
+    }
+
+    pub fn expanded_display(&self) -> String {
+        format!(
+            "Topic to Topic Sync Process: {} -> {}",
+            self.source_topic_id, self.target_topic_id
+        )
+    }
+
+    pub fn short_display(&self) -> String {
+        format!(
+            "Topic to Topic Sync Process: {} -> {}",
+            self.source_topic_id, self.target_topic_id
+        )
+    }
+}

apps/framework-cli/src/framework/core/infrastructure_map.rs

@@ -4,7 +4,7 @@ use super::infrastructure::function_process::FunctionProcess;
 use super::infrastructure::olap_process::OlapProcess;
 use super::infrastructure::table::Table;
 use super::infrastructure::topic::Topic;
-use super::infrastructure::topic_to_table_sync_process::TopicToTableSyncProcess;
+use super::infrastructure::topic_sync_process::{TopicToTableSyncProcess, TopicToTopicSyncProcess};
 use super::infrastructure::view::View;
 use super::primitive_map::PrimitiveMap;
 use crate::framework::controller::{InitialDataLoad, InitialDataLoadStatus};
@@ -60,6 +60,7 @@ pub enum ApiChange {
 #[derive(Debug, Clone)]
 pub enum ProcessChange {
     TopicToTableSyncProcess(Change<TopicToTableSyncProcess>),
+    TopicToTopicSyncProcess(Change<TopicToTopicSyncProcess>),
     FunctionProcess(Change<FunctionProcess>),
     OlapProcess(Change<OlapProcess>),
     ConsumptionApiWebServer(Change<ConsumptionApiWebServer>),
@@ -101,6 +102,8 @@ pub struct InfrastructureMap {
     pub views: HashMap<String, View>,
 
     pub topic_to_table_sync_processes: HashMap<String, TopicToTableSyncProcess>,
+    #[serde(default = "HashMap::new")]
+    pub topic_to_topic_sync_processes: HashMap<String, TopicToTopicSyncProcess>,
     pub function_processes: HashMap<String, FunctionProcess>,
 
     // TODO change to a hashmap of processes when we have several
@@ -120,6 +123,7 @@ impl InfrastructureMap {
         let mut topics = HashMap::new();
         let mut api_endpoints = HashMap::new();
         let mut topic_to_table_sync_processes = HashMap::new();
+        let mut topic_to_topic_sync_processes = HashMap::new();
         let mut function_processes = HashMap::new();
         let mut initial_data_loads = HashMap::new();
 
@@ -208,6 +212,9 @@ impl InfrastructureMap {
                 );
                 topic_to_table_sync_processes.insert(sync_process.id(), sync_process);
 
+                let topic_sync = TopicToTopicSyncProcess::from_migration_function(function);
+                topic_to_topic_sync_processes.insert(topic_sync.id(), topic_sync);
+
                 initial_data_loads.insert(
                     function_process.id(),
                     InitialDataLoad {
@@ -240,6 +247,7 @@ impl InfrastructureMap {
             topics,
             api_endpoints,
             topic_to_table_sync_processes,
+            topic_to_topic_sync_processes,
             tables,
             views,
             function_processes,
@@ -421,6 +429,47 @@ impl InfrastructureMap {
             }
         }
 
+        // =================================================================
+        //                              Topic to Topic Sync Processes
+        // =================================================================
+
+        for (id, topic_to_topic_sync_process) in &self.topic_to_topic_sync_processes {
+            if let Some(target_topic_to_topic_sync_process) =
+                target_map.topic_to_topic_sync_processes.get(id)
+            {
+                if topic_to_topic_sync_process != target_topic_to_topic_sync_process {
+                    changes
+                        .processes_changes
+                        .push(ProcessChange::TopicToTopicSyncProcess(Change::<
+                            TopicToTopicSyncProcess,
+                        >::Updated {
+                            before: topic_to_topic_sync_process.clone(),
+                            after: target_topic_to_topic_sync_process.clone(),
+                        }));
+                }
+            } else {
+                changes
+                    .processes_changes
+                    .push(ProcessChange::TopicToTopicSyncProcess(Change::<
+                        TopicToTopicSyncProcess,
+                    >::Removed(
+                        topic_to_topic_sync_process.clone(),
+                    )));
+            }
+        }
+
+        for (id, topic_to_topic_sync_process) in &target_map.topic_to_topic_sync_processes {
+            if !self.topic_to_topic_sync_processes.contains_key(id) {
+                changes
+                    .processes_changes
+                    .push(ProcessChange::TopicToTopicSyncProcess(Change::<
+                        TopicToTopicSyncProcess,
+                    >::Added(
+                        topic_to_topic_sync_process.clone(),
+                    )));
+            }
+        }
+
         // =================================================================
         //                             Function Processes
         // =================================================================
@@ -574,6 +623,17 @@ impl InfrastructureMap {
             })
             .collect();
 
+        let mut topic_to_topic_process_changes: Vec<ProcessChange> = self
+            .topic_to_topic_sync_processes
+            .values()
+            .map(|topic_to_table_sync_process| {
+                ProcessChange::TopicToTopicSyncProcess(Change::<TopicToTopicSyncProcess>::Added(
+                    topic_to_table_sync_process.clone(),
+                ))
+            })
+            .collect();
+        topic_to_table_process_changes.append(&mut topic_to_topic_process_changes);
+
         let mut function_process_changes: Vec<ProcessChange> = self
             .function_processes
             .values()

apps/framework-cli/src/infrastructure/olap/clickhouse_alt_client.rs

@@ -305,7 +305,7 @@ pub async fn retrieve_current_state(
 #[derive(Debug, thiserror::Error)]
 #[non_exhaustive]
 pub enum StateStorageError {
-    #[error("Failed to serialize the state")]
+    #[error("Failed to (de)serialize the state")]
     SerdeError(#[from] serde_json::Error),
     #[error("Clickhouse error")]
     ClickhouseError(#[from] clickhouse_rs::errors::Error),

apps/framework-cli/src/infrastructure/processes.rs

@@ -74,6 +74,28 @@ pub async fn execute_changes(
                     metrics.clone(),
                 );
             }
+            ProcessChange::TopicToTopicSyncProcess(Change::Added(sync)) => {
+                log::info!("Starting sync process: {:?}", sync.id());
+                syncing_registry.start_topic_to_topic(
+                    sync.source_topic_id.clone(),
+                    sync.target_topic_id.clone(),
+                    metrics.clone(),
+                );
+            }
+            ProcessChange::TopicToTopicSyncProcess(Change::Removed(sync)) => {
+                log::info!("Stopping sync process: {:?}", sync.id());
+                syncing_registry.stop_topic_to_topic(&sync.target_topic_id)
+            }
+            // TopicToTopicSyncProcess Updated seems impossible
+            ProcessChange::TopicToTopicSyncProcess(Change::Updated { before, after }) => {
+                log::info!("Replacing Sync process: {:?} by {:?}", before, after);
+                syncing_registry.stop_topic_to_topic(&before.target_topic_id);
+                syncing_registry.start_topic_to_topic(
+                    after.source_topic_id.clone(),
+                    after.target_topic_id.clone(),
+                    metrics.clone(),
+                );
+            }
             ProcessChange::FunctionProcess(Change::Added(function_process)) => {
                 log::info!("Starting Function process: {:?}", function_process.id());
                 process_registry.functions.start(function_process)?;

apps/framework-cli/src/infrastructure/processes/kafka_clickhouse_sync.rs

@@ -204,6 +204,36 @@ impl SyncingProcessesRegistry {
             process.abort();
         }
     }
+
+    pub fn start_topic_to_topic(
+        &mut self,
+        source_topic_name: String,
+        target_topic_name: String,
+        metrics: Arc<Metrics>,
+    ) {
+        info!(
+            "<DCM> Starting syncing process from topic: {} to topic: {}",
+            source_topic_name, target_topic_name
+        );
+        let key = target_topic_name.clone();
+
+        if let Some(process) = self.to_topic_registry.remove(&key) {
+            process.abort();
+        }
+
+        self.insert_topic_sync(spawn_kafka_to_kafka_process(
+            self.kafka_config.clone(),
+            source_topic_name,
+            target_topic_name,
+            metrics.clone(),
+        ));
+    }
+
+    pub fn stop_topic_to_topic(&mut self, target_topic_name: &str) {
+        if let Some(process) = self.to_table_registry.remove(target_topic_name) {
+            process.abort();
+        }
+    }
 }
 
 type FnSyncProcess =