Test KafkaClient::commit_offset

Author: dead10ck

src/utils.rs

@@ -4,7 +4,7 @@
 
 /// A retrieved offset for a particular partition in the context of an
 /// already known topic.
-#[derive(Debug)]
+#[derive(Debug, Hash, PartialEq, Eq)]
 pub struct PartitionOffset {
     pub offset: i64,
     pub partition: i32,

tests/integration/client/mod.rs

@@ -6,7 +6,7 @@
 use super::*;
 use std::collections::HashSet;
 use std::time::Duration;
-use kafka::client::{KafkaClient, FetchPartition, ProduceMessage, RequiredAcks};
+use kafka::client::{KafkaClient, PartitionOffset, FetchPartition, ProduceMessage, RequiredAcks};
 use kafka::client::fetch::Response;
 
 fn flatten_fetched_messages(resps: &Vec<Response>) -> Vec<(&str, i32, &[u8])> {
@@ -65,9 +65,7 @@ fn test_kafka_client_load_metadata() {
 
 #[test]
 fn test_produce_fetch_messages() {
-    let hosts = vec![LOCAL_KAFKA_BOOTSTRAP_HOST.to_owned()];
-    let mut client = KafkaClient::new(hosts.clone());
-    client.load_metadata_all().unwrap();
+    let mut client = new_ready_kafka_client();
 
     // first send the messages and verify correct confirmation responses
     // from kafka
@@ -124,3 +122,35 @@ fn test_produce_fetch_messages() {
                 .into_iter()
                 .all(|c_msg| messages.contains(&c_msg)));
 }
+
+#[test]
+fn test_commit_offset() {
+    let mut client = new_ready_kafka_client();
+
+    for &(partition, offset) in
+        &[(TEST_TOPIC_PARTITIONS[0], 100),
+          (TEST_TOPIC_PARTITIONS[1], 200),
+          (TEST_TOPIC_PARTITIONS[0], 300),
+          (TEST_TOPIC_PARTITIONS[1], 400),
+          (TEST_TOPIC_PARTITIONS[0], 500),
+          (TEST_TOPIC_PARTITIONS[1], 600)] {
+
+
+        client
+            .commit_offset(TEST_GROUP_NAME, TEST_TOPIC_NAME, partition, offset)
+            .unwrap();
+
+        let partition_offsets: HashSet<PartitionOffset> = client
+            .fetch_group_topic_offsets(TEST_GROUP_NAME, TEST_TOPIC_NAME)
+            .unwrap()
+            .into_iter()
+            .collect();
+
+        let correct_partition_offset = PartitionOffset {
+            partition: partition,
+            offset: offset,
+        };
+
+        assert!(partition_offsets.contains(&correct_partition_offset));
+    }
+}

tests/test_kafka.rs

@@ -3,6 +3,8 @@ extern crate kafka;
 
 #[cfg(feature = "integration_tests")]
 mod integration {
+    use kafka::client::KafkaClient;
+
     mod client;
     mod consumer_producer;
 
@@ -12,4 +14,11 @@ mod integration {
     pub const TEST_GROUP_NAME: &str = "kafka-rust-tester";
     pub const TEST_TOPIC_PARTITIONS: [i32; 2] = [0, 1];
     pub const KAFKA_CONSUMER_OFFSETS_TOPIC_NAME: &str = "__consumer_offsets";
+
+    pub(crate) fn new_ready_kafka_client() -> KafkaClient {
+        let hosts = vec![LOCAL_KAFKA_BOOTSTRAP_HOST.to_owned()];
+        let mut client = KafkaClient::new(hosts);
+        client.load_metadata_all().unwrap();
+        client
+    }
 }