fix log

Author: luoluoyuyu

src/logging/mod.rs

@@ -50,12 +50,7 @@ pub fn init_logging(config: &LogConfig) -> Result<()> {
 
     let subscriber = Registry::default()
         .with(log_level)
-        .with(
-            fmt::layer()
-                .with_writer(non_blocking)
-                .with_ansi(false)
-                .json(),
-        )
+        .with(fmt::layer().with_writer(non_blocking).with_ansi(false))
         .with(fmt::layer().with_writer(std::io::stdout).with_ansi(true));
 
     subscriber.init();

src/runtime/output/protocol/kafka/output_sink.rs

@@ -474,14 +474,6 @@ impl KafkaOutputSink {
     ) {
         // Use into_buffer() to take ownership, avoids extra copy
         if let Some(payload) = data.into_buffer() {
-            let payload_str = String::from_utf8_lossy(&payload);
-            log::info!(
-                "Sending to Kafka topic '{}': len={}, payload={}",
-                config.topic,
-                payload.len(),
-                payload_str
-            );
-
             let mut record: BaseRecord<'_, (), Vec<u8>> =
                 BaseRecord::to(&config.topic).payload(&payload);
 
@@ -742,17 +734,6 @@ impl OutputSink for KafkaOutputSink {
     // -------------------- collect --------------------
 
     fn collect(&mut self, data: BufferOrEvent) -> Result<(), Box<dyn std::error::Error + Send>> {
-        // Print current state
-        let state = self.state.lock().unwrap().clone();
-        let data_sender_exists = self.data_sender.is_some();
-        log::info!(
-            "KafkaOutputSink collect: sink_id={}, topic={}, state={:?}, data_sender_exists={}",
-            self.sink_id,
-            self.config.topic,
-            state,
-            data_sender_exists
-        );
-
         // Don't check state in main thread, send data directly to channel
         // If runloop is not running, data will be queued in channel
         if let Some(ref sender) = self.data_sender {

src/runtime/processor/wasm/wasm_processor.rs

@@ -241,17 +241,6 @@ impl WasmProcessor for WasmProcessorImpl {
             ))
         })?;
 
-        // Call wasm process function
-        // WIT: export fs-process: func(source-id: u32, data: list<u8>);
-        let payload_str = String::from_utf8_lossy(&data);
-        log::info!(
-            "Calling fs_process: input_index={}, data_len={}, payload={}",
-            input_index,
-            data.len(),
-            payload_str
-        );
-
-        let start = std::time::Instant::now();
         processor
             .call_fs_process(store, input_index as u32, &data)
             .map_err(|e| -> Box<dyn Error + Send> {
@@ -260,12 +249,6 @@ impl WasmProcessor for WasmProcessorImpl {
                     e
                 )))
             })?;
-        let elapsed_us = start.elapsed().as_micros();
-        log::info!(
-            "fs_process completed: input_index={}, elapsed={}us",
-            input_index,
-            elapsed_us
-        );
 
         log::debug!(
             "WasmProcessor '{}' processed {} bytes from input {}",