diff --git a/agent/src/flow_generator/protocol_logs/mq/kafka.rs b/agent/src/flow_generator/protocol_logs/mq/kafka.rs index 0cbc969fc29..d4f582108d3 100644 --- a/agent/src/flow_generator/protocol_logs/mq/kafka.rs +++ b/agent/src/flow_generator/protocol_logs/mq/kafka.rs @@ -440,6 +440,7 @@ impl KafkaLog { const MSG_LEN_SIZE: usize = 4; const MAX_TRACE_ID: usize = 255; const MAX_SESSION_PER_FLOW: usize = 32; + const MAX_VERSION: u16 = 12; fn decode_varint(buf: &[u8]) -> (usize, usize) { let mut shift = 0; @@ -1496,6 +1497,9 @@ impl KafkaLog { info.msg_type = LogMessageType::Request; info.api_key = read_u16_be(&payload[4..]); info.api_version = read_u16_be(&payload[6..]); + if info.api_version > Self::MAX_VERSION { + return Err(Error::KafkaLogParseFailed); + } info.correlation_id = read_u32_be(&payload[8..]); info.client_id = String::from_utf8_lossy(&payload[14..14 + client_id_len]).into_owned(); if !info.client_id.is_ascii() { diff --git a/agent/src/rpc/synchronizer.rs b/agent/src/rpc/synchronizer.rs index 09c7d53e03b..7930cf4a572 100644 --- a/agent/src/rpc/synchronizer.rs +++ b/agent/src/rpc/synchronizer.rs @@ -1256,7 +1256,7 @@ impl Synchronizer { return; } ntp_diff.store(0, Ordering::Relaxed); - time::sleep(sync_interval).await; + time::sleep(Duration::from_secs(1)).await; continue; }