Test checking close waits

Author: tharinduamila-insta

shotover-proxy/tests/kafka_int_tests/mod.rs

@@ -350,6 +350,11 @@ async fn cluster_1_rack_single_shotover_broker_idle_timeout(#[case] driver: Kafk
     // So instead we rely on a test case hits the timeout with plenty of buffer to avoid the race condition.
     test_cases::test_broker_idle_timeout(&connection_builder).await;
 
+    //we had an issue where shotover was not acknowledging the closed connecitons from Kafka
+    //leaving around close-wait state connections. This step ensures no such connections are left behind
+    //for the given destination port and owned by the shotover process 
+    test_cases::assert_no_close_wait_connections(shotover.pid());
+
     tokio::time::timeout(
         Duration::from_secs(10),
         shotover.shutdown_and_then_consume_events(&[]),

shotover-proxy/tests/kafka_int_tests/test_cases.rs

@@ -2100,3 +2100,45 @@ pub async fn test_no_out_of_rack_request(connection_builder: &KafkaConnectionBui
         0
     );
 }
+
+/// Counts CLOSE_WAIT connections for a given PID using lsof
+pub fn count_close_wait_connections_for_process(pid: i32) -> anyhow::Result<usize> {
+    use test_helpers::run_command;
+    
+    // Use lsof to get network connections for the specific PID
+    // -i: network files
+    // -P: don't resolve port numbers to names
+    // -n: don't resolve hostnames
+    // -a: AND the conditions together
+    let output = run_command(
+        "lsof",
+        &["-i", "-P", "-n", "-a", "-p", &pid.to_string()],
+    )?;
+
+    let mut count = 0;
+    
+    for line in output.lines() {
+        // lsof output format (columns vary, but typically):
+        // COMMAND  PID  USER  FD  TYPE  DEVICE  SIZE/OFF  NODE  NAME
+        // The NAME column contains the connection info like "127.0.0.1:54321->172.16.1.2:9092 (CLOSE_WAIT)"
+        
+        // Check if line contains CLOSE_WAIT state
+        if line.contains("CLOSE_WAIT") {
+            count += 1;
+        }
+    }
+    Ok(count)
+}
+
+/// Asserts that there are no CLOSE_WAIT connections for the given PID and destination port
+pub fn assert_no_close_wait_connections(pid: i32) {
+    let count = count_close_wait_connections_for_process(pid)
+        .expect("Failed to check for CLOSE_WAIT connections");
+    
+    assert_eq!(
+        count, 0,
+        "Found {} CLOSE_WAIT connection(s) for PID {}. \
+         This indicates connections that were closed by the remote peer but not yet closed by the local process.",
+        count, pid
+    );
+}