include broker id in log string

shotover · Nov 19, 2024 · 4a1c9fc · 4a1c9fc
1 parent 39f99df
commit 4a1c9fc
Show file tree

Hide file tree

Showing 2 changed files with 136 additions and 77 deletions.
diff --git a/shotover-proxy/tests/kafka_int_tests/test_cases.rs b/shotover-proxy/tests/kafka_int_tests/test_cases.rs
@@ -1853,40 +1853,52 @@ pub async fn describe_log_dirs(connection_builder: &KafkaConnectionBuilder) {
             },
         ])
         .await;
-    assert_eq!(
-        result,
-        HashMap::from([
-            (
-                TopicPartitionReplica {
-                    topic_name: "describe_logs_test".to_owned(),
-                    partition: 0,
-                    broker_id: 0,
-                },
-                DescribeReplicaLogDirInfo {
-                    path: Some("/bitnami/kafka/data".to_owned())
-                }
-            ),
-            (
-                TopicPartitionReplica {
-                    topic_name: "describe_logs_test".to_owned(),
-                    partition: 0,
-                    broker_id: 1,
-                },
-                DescribeReplicaLogDirInfo {
-                    path: Some("/bitnami/kafka/data".to_owned())
-                }
-            ),
-            (
-                TopicPartitionReplica {
-                    topic_name: "describe_logs_test2".to_owned(),
-                    partition: 0,
-                    broker_id: 0,
-                },
-                DescribeReplicaLogDirInfo {
-                    path: Some("/bitnami/kafka/data".to_owned())
-                }
-            )
-        ])
+
+    /// Assert that the path in the DescribeLogsDir response matches the custom format used by shotover.
+    /// This format looks like: actual-kafka-broker-id3:/original/kafka/path/here
+    fn assert_valid_path(info: &DescribeReplicaLogDirInfo) {
+        let id = info
+            .path
+            .as_ref()
+            .unwrap()
+            .strip_prefix("actual-kafka-broker-id")
+            .unwrap()
+            .strip_suffix(":/bitnami/kafka/data")
+            .unwrap();
+        let id: i32 = id.parse().unwrap();
+        assert!(
+            id < 6,
+            "There are only 6 brokers so the broker id must be between 0-5 inclusive"
+        );
+    }
+
+    assert_eq!(result.len(), 3);
+    assert_valid_path(
+        result
+            .get(&TopicPartitionReplica {
+                topic_name: "describe_logs_test".to_owned(),
+                partition: 0,
+                broker_id: 0,
+            })
+            .unwrap(),
+    );
+    assert_valid_path(
+        result
+            .get(&TopicPartitionReplica {
+                topic_name: "describe_logs_test".to_owned(),
+                partition: 0,
+                broker_id: 1,
+            })
+            .unwrap(),
+    );
+    assert_valid_path(
+        result
+            .get(&TopicPartitionReplica {
+                topic_name: "describe_logs_test2".to_owned(),
+                partition: 0,
+                broker_id: 0,
+            })
+            .unwrap(),
     );
 }