From 4ee9e61c8287314558179956fcadd6fa4c4166b9 Mon Sep 17 00:00:00 2001
From: Lucas Kent <rubickent@gmail.com>
Date: Thu, 17 Oct 2024 16:12:46 +1100
Subject: [PATCH] Improve consumer assertion

---
 test-helpers/src/connection/kafka/mod.rs | 70 +++++++++++++++++-------
 1 file changed, 51 insertions(+), 19 deletions(-)

diff --git a/test-helpers/src/connection/kafka/mod.rs b/test-helpers/src/connection/kafka/mod.rs
index 99d3e91ca..3d6c876f9 100644
--- a/test-helpers/src/connection/kafka/mod.rs
+++ b/test-helpers/src/connection/kafka/mod.rs
@@ -200,25 +200,57 @@ impl KafkaConsumer {
             Self::Java(java) => java.consume().await,
         };
 
-        let topic = &expected_response.topic_name;
-        assert_eq!(
-            expected_response.topic_name, response.topic_name,
-            "Unexpected topic"
-        );
-        assert_eq!(
-            expected_response.message, response.message,
-            "Unexpected message for topic {topic}"
-        );
-        assert_eq!(
-            expected_response.key, response.key,
-            "Unexpected key for topic {topic}"
-        );
-
-        if expected_response.offset.is_some() {
-            assert_eq!(
-                expected_response.offset, response.offset,
-                "Unexpected offset for topic {topic}"
-            );
+        // Construct an error message that gives as much context as possible as to what went wrong.
+        let mut error = false;
+
+        let expected_topic = &expected_response.topic_name;
+        let actual_topic = &response.topic_name;
+        let topic_result = if expected_topic == actual_topic {
+            "️and it matched".into()
+        } else {
+            error = true;
+            format!("but the topic was {actual_topic:?}")
+        };
+
+        let expected_message = &expected_response.message;
+        let actual_message = &response.message;
+        let message_result = if expected_message == actual_message {
+            "and it matched".into()
+        } else {
+            error = true;
+            format!("but the message was {actual_message:?}")
+        };
+
+        let expected_key = &expected_response.key;
+        let actual_key = &response.key;
+        let key_result = if expected_key == actual_key {
+            "and it matched".into()
+        } else {
+            error = true;
+            format!("but the key was {actual_key:?}")
+        };
+
+        let expected_offset = &expected_response.offset;
+        let actual_offset = &response.offset;
+        let offset_result = if expected_offset.is_some() {
+            if expected_offset == actual_offset {
+                format!("expected offset {expected_offset:?} and it matched")
+            } else {
+                error = true;
+                format!("expected offset {expected_offset:?} but the offset was {actual_offset:?}")
+            }
+        } else {
+            format!("No offset expected and the offset was {actual_offset:?}")
+        };
+
+        if error {
+            panic!(
+                r#"Consumed an unexpected record:
+  expected topic {expected_topic:?} {topic_result}
+  expected message {expected_message:?} {message_result}
+  expected key {expected_key:?} {key_result}
+  {offset_result}"#
+            )
         }
     }