feat(kafka): kafka protocol support.

Author: echooymxq

Cargo.lock

@@ -302,7 +302,7 @@ incremental = true           # Enable incremental compilation (auto-cleaned peri
 opt-level = 0                # No optimization (fastest compile time)
 split-debuginfo = "unpacked" # Reduce disk usage on macOS/Linux
 codegen-units = 16           # Balanced parallelism (was 256, reduced to prevent cache explosion)
-debug = 1                    # Reduced debug info (line tables only) - faster linking
+debug = 2
 
 # Optimize dependencies even in dev mode for better runtime performance
 [profile.dev.package."*"]

Cargo.toml

@@ -25,6 +25,7 @@ axum.workspace = true
 thiserror.workspace = true
 mqtt-broker.workspace = true
 meta-service.workspace = true
+kafka-broker.workspace = true
 tonic.workspace = true
 tower-http = { workspace = true, features = ["cors"] }
 tonic-web.workspace = true

src/broker-server/Cargo.toml

@@ -31,6 +31,7 @@ use journal_server::{
     server::connection_manager::ConnectionManager as JournalConnectionManager, JournalServer,
     JournalServerParams,
 };
+use kafka_broker::broker::{KafkaBrokerServer, KafkaBrokerServerParams};
 use meta_service::{
     controller::{
         journal::call_node::JournalInnerCallManager, mqtt::call_broker::MQTTInnerCallManager,
@@ -47,6 +48,7 @@ use mqtt_broker::{
     subscribe::manager::SubscribeManager,
 };
 use network_server::common::connection_manager::ConnectionManager as NetworkConnectionManager;
+use network_server::context::ProcessorConfig;
 use pprof_monitor::pprof_monitor::start_pprof_monitor;
 use rate_limit::RateLimiterManager;
 use rocksdb_engine::{
@@ -80,6 +82,7 @@ pub struct BrokerServer {
     main_runtime: Runtime,
     place_params: MetaServiceServerParams,
     mqtt_params: MqttBrokerServerParams,
+    kafka_params: KafkaBrokerServerParams,
     journal_params: JournalServerParams,
     client_pool: Arc<ClientPool>,
     rocksdb_engine_handler: Arc<RocksDBEngine>,
@@ -144,6 +147,12 @@ impl BrokerServer {
             message_storage_adapter,
         );
 
+        let kafka_params = BrokerServer::build_kafka_server_params(
+            client_pool.clone(),
+            broker_cache.clone(),
+            connection_manager.clone(),
+        );
+
         let journal_params = BrokerServer::build_journal_server(client_pool.clone());
 
         BrokerServer {
@@ -153,6 +162,7 @@ impl BrokerServer {
             journal_params,
             config: config.clone(),
             mqtt_params,
+            kafka_params,
             client_pool,
             rocksdb_engine_handler,
             rate_limiter_manager,
@@ -226,6 +236,7 @@ impl BrokerServer {
         let mut place_stop_send = None;
         let mut mqtt_stop_send = None;
         let mut journal_stop_send = None;
+        let mut kafka_stop_send = None;
 
         let config = broker_config();
         // start meta service
@@ -266,12 +277,20 @@ impl BrokerServer {
         let (stop_send, _) = broadcast::channel(2);
         let mqtt_runtime =
             create_runtime("mqtt-runtime", self.config.runtime.runtime_worker_threads);
+        let kafka_runtime =
+            create_runtime("kafka-runtime", self.config.runtime.runtime_worker_threads);
         if config.is_start_broker() {
             mqtt_stop_send = Some(stop_send.clone());
             let server = MqttBrokerServer::new(self.mqtt_params.clone(), stop_send.clone());
             mqtt_runtime.spawn(async move {
                 server.start().await;
             });
+
+            kafka_stop_send = Some(stop_send.clone());
+            let kafka_server = KafkaBrokerServer::new(self.kafka_params.clone(), stop_send.clone());
+            kafka_runtime.spawn(async move {
+                kafka_server.start().await;
+            });
         }
 
         // register node
@@ -293,7 +312,12 @@ impl BrokerServer {
         });
 
         // awaiting stop
-        self.awaiting_stop(place_stop_send, mqtt_stop_send, journal_stop_send);
+        self.awaiting_stop(
+            place_stop_send,
+            mqtt_stop_send,
+            journal_stop_send,
+            kafka_stop_send,
+        );
     }
 
     async fn build_meta_service(
@@ -376,6 +400,26 @@ impl BrokerServer {
         }
     }
 
+    fn build_kafka_server_params(
+        client_pool: Arc<ClientPool>,
+        broker_cache: Arc<BrokerCacheManager>,
+        connection_manager: Arc<NetworkConnectionManager>,
+    ) -> KafkaBrokerServerParams {
+        let conf = broker_config();
+        let proc_config = ProcessorConfig {
+            accept_thread_num: conf.network.accept_thread_num,
+            handler_process_num: conf.network.handler_thread_num,
+            response_process_num: conf.network.response_thread_num,
+            channel_size: conf.network.queue_size,
+        };
+        KafkaBrokerServerParams {
+            connection_manager,
+            client_pool,
+            proc_config,
+            broker_cache,
+        }
+    }
+
     fn build_journal_server(client_pool: Arc<ClientPool>) -> JournalServerParams {
         let config = broker_config();
         let connection_manager = Arc::new(JournalConnectionManager::new());
@@ -403,6 +447,7 @@ impl BrokerServer {
         place_stop: Option<broadcast::Sender<bool>>,
         mqtt_stop: Option<broadcast::Sender<bool>>,
         journal_stop: Option<broadcast::Sender<bool>>,
+        kafka_stop: Option<broadcast::Sender<bool>>,
     ) {
         self.main_runtime.block_on(async {
             self.broker_cache
@@ -426,6 +471,13 @@ impl BrokerServer {
                 sleep(Duration::from_secs(3));
             }
 
+            if let Some(sx) = kafka_stop {
+                if let Err(e) = sx.send(true) {
+                    error!("kafka stop signal, error message:{}", e);
+                }
+                sleep(Duration::from_secs(3));
+            }
+
             if let Some(sx) = journal_stop {
                 if let Err(e) = sx.send(true) {
                     error!("journal stop signal, error message{}", e);

src/broker-server/src/lib.rs

@@ -15,9 +15,9 @@
 use super::default::{
     default_broker_id, default_cluster_name, default_flapping_detect, default_grpc_port,
     default_http_port, default_journal_runtime, default_journal_server, default_journal_storage,
-    default_message_storage, default_meta_addrs, default_mqtt_auth_config, default_mqtt_keep_alive,
-    default_mqtt_offline_message, default_mqtt_protocol_config, default_mqtt_runtime,
-    default_mqtt_schema, default_mqtt_security, default_mqtt_server,
+    default_kafka_config, default_message_storage, default_meta_addrs, default_mqtt_auth_config,
+    default_mqtt_keep_alive, default_mqtt_offline_message, default_mqtt_protocol_config,
+    default_mqtt_runtime, default_mqtt_schema, default_mqtt_security, default_mqtt_server,
     default_mqtt_slow_subscribe_config, default_mqtt_system_monitor, default_network,
     default_place_runtime, default_rocksdb, default_roles, default_runtime,
 };
@@ -122,6 +122,10 @@ pub struct BrokerConfig {
 
     #[serde(default = "default_mqtt_system_monitor")]
     pub mqtt_system_monitor: MqttSystemMonitor,
+
+    // KAFKA
+    #[serde(default = "default_kafka_config")]
+    pub kafka_config: KafkaConfig,
 }
 
 impl BrokerConfig {
@@ -352,3 +356,8 @@ pub struct JournalStorage {
 pub struct JournalServer {
     pub tcp_port: u32,
 }
+
+#[derive(Serialize, Deserialize, Default, Clone, Debug)]
+pub struct KafkaConfig {
+    pub port: u32,
+}

src/common/config/src/config.rs

@@ -14,10 +14,10 @@
 
 use super::security::{AuthnConfig, AuthzConfig};
 use crate::config::{
-    JournalRuntime, JournalServer, JournalStorage, MetaRuntime, MqttAuthConfig, MqttFlappingDetect,
-    MqttKeepAlive, MqttOfflineMessage, MqttProtocolConfig, MqttRuntime, MqttSchema, MqttSecurity,
-    MqttServer, MqttSlowSubscribeConfig, MqttSystemMonitor, Network, Rocksdb, Runtime,
-    SchemaFailedOperation, SchemaStrategy,
+    JournalRuntime, JournalServer, JournalStorage, KafkaConfig, MetaRuntime, MqttAuthConfig,
+    MqttFlappingDetect, MqttKeepAlive, MqttOfflineMessage, MqttProtocolConfig, MqttRuntime,
+    MqttSchema, MqttSecurity, MqttServer, MqttSlowSubscribeConfig, MqttSystemMonitor, Network,
+    Rocksdb, Runtime, SchemaFailedOperation, SchemaStrategy,
 };
 use crate::storage::{StorageAdapterConfig, StorageAdapterType};
 use common_base::enum_type::delay_type::DelayType;
@@ -208,3 +208,7 @@ pub fn default_journal_storage() -> JournalStorage {
         rocksdb_max_open_files: 10000,
     }
 }
+
+pub fn default_kafka_config() -> KafkaConfig {
+    KafkaConfig { port: 9092 }
+}

src/common/config/src/default.rs

@@ -21,8 +21,10 @@ use common_base::tools::now_second;
 use dashmap::DashMap;
 use futures::stream::SplitSink;
 use futures::SinkExt;
+use kafka_protocol::messages::ResponseHeader;
 use metadata_struct::connection::{NetworkConnection, NetworkConnectionType};
 use protocol::codec::{RobustMQCodec, RobustMQCodecWrapper};
+use protocol::kafka::packet::{KafkaHeader, KafkaPacketWrapper};
 use protocol::mqtt::codec::MqttPacketWrapper;
 use protocol::robust::{RobustMQPacket, RobustMQPacketWrapper, RobustMQProtocol};
 use std::time::Duration;
@@ -219,18 +221,31 @@ impl ConnectionManager {
         };
 
         if packet_wrapper.protocol.is_mqtt() {
-            if let RobustMQPacket::MQTT(pack) = packet_wrapper.packet {
+            if let RobustMQPacket::MQTT(pack) = &packet_wrapper.packet {
                 let mqtt_packet = MqttPacketWrapper {
                     protocol_version: packet_wrapper.protocol.to_u8(),
-                    packet: pack,
+                    packet: pack.clone(),
                 };
                 self.write_mqtt_tcp_frame(connection_id, mqtt_packet)
                     .await?;
             }
         }
 
         if packet_wrapper.protocol.is_kafka() {
-            // todo
+            if let RobustMQPacket::KAFKA(pack) = &packet_wrapper.packet {
+                let request_header = packet_wrapper.extend.get_kafka_header();
+                let kafka_packet = KafkaPacketWrapper {
+                    api_key: request_header.request_api_key,
+                    api_version: request_header.request_api_version,
+                    header: KafkaHeader::Response(
+                        ResponseHeader::default()
+                            .with_correlation_id(request_header.correlation_id),
+                    ),
+                    packet: pack.clone(),
+                };
+                self.write_kafka_tcp_frame(connection_id, kafka_packet)
+                    .await?;
+            }
         }
         Ok(())
     }
@@ -486,4 +501,47 @@ impl ConnectionManager {
             };
         }
     }
+
+    pub async fn write_kafka_tcp_frame(
+        &self,
+        connection_id: u64,
+        pack: KafkaPacketWrapper,
+    ) -> ResultCommonError {
+        let mut times = 0;
+        loop {
+            match self.tcp_write_list.try_get_mut(&connection_id) {
+                dashmap::try_result::TryResult::Present(mut da) => {
+                    match da.send(RobustMQCodecWrapper::KAFKA(pack.clone())).await {
+                        Ok(_) => {
+                            break;
+                        }
+                        Err(e) => {
+                            if broker_not_available(&e.to_string()) {
+                                return Err(CommonError::CommonError(e.to_string()));
+                            }
+
+                            if times > self.lock_max_try_mut_times {
+                                return Err(CommonError::FailedToWriteClient(
+                                    "tcp".to_string(),
+                                    e.to_string(),
+                                ));
+                            }
+                        }
+                    }
+                }
+                dashmap::try_result::TryResult::Absent => {
+                    if times > self.lock_max_try_mut_times {
+                        return Err(CommonError::NotObtainAvailableConnection(
+                            "tcp".to_string(),
+                            connection_id,
+                        ));
+                    }
+                }
+                dashmap::try_result::TryResult::Locked => {}
+            }
+            times += 1;
+            sleep(Duration::from_millis(self.lock_try_mut_sleep_time_ms)).await
+        }
+        Ok(())
+    }
 }