Repository: kafka-rust/kafka-rust (language: rust)

Issue:
SipHasher has been deprecated
[`SipHasher`](https://doc.rust-lang.org/std/hash/struct.SipHasher.html) has been deprecated in favor of [`DefaultHasher`](https://doc.rust-lang.org/std/collections/hash_map/struct.DefaultHasher.html), but `DefaultHasher` says

> The internal algorithm is not specified, and so it and its hashes should not be relied upon over releases.

We need to use a stable hashing algorithm to avoid surprises with data distribution.


Discussion / hints:
just for reference; the `KafkaProducer` part of the official kafka distribution uses a [32bit murmur2 hash](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java#L69) by default.

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index 2d343f461..8bc17e23d 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -17,6 +17,7 @@ crc = "1.3"
 log = "0.3"
 ref_slice = "1.0"
 fnv = "1.0"
+twox-hash = "1.1"
 
 flate2 = { version = "0.2", optional = true }
 snappy = { version = "0.4", optional = true }
diff --git a/src/lib.rs b/src/lib.rs
index 9c2052b56..2aa44822b 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -14,6 +14,7 @@ extern crate byteorder;
 extern crate crc;
 extern crate ref_slice;
 extern crate fnv;
+extern crate twox_hash;
 
 #[macro_use]
 extern crate log;
diff --git a/src/producer.rs b/src/producer.rs
index 7de3c751e..d6b87c6a8 100644
--- a/src/producer.rs
+++ b/src/producer.rs
@@ -61,11 +61,12 @@
 
 use std::collections::HashMap;
 use std::fmt;
-use std::hash::{Hasher, SipHasher, BuildHasher, BuildHasherDefault};
+use std::hash::{Hasher, BuildHasher, BuildHasherDefault};
 use std::time::Duration;
 use client::{self, KafkaClient};
 use error::{Error, Result};
 use ref_slice::ref_slice;
+use twox_hash::XxHash32;
 
 #[cfg(feature = "security")]
 use client::SecurityConfig;
@@ -546,6 +547,9 @@ pub trait Partitioner {
     fn partition(&mut self, topics: Topics, msg: &mut client::ProduceMessage);
 }
 
+/// The default hasher implementation used of `DefaultPartitioner`.
+pub type DefaultHasher = XxHash32;
+
 /// As its name implies `DefaultPartitioner` is the default
 /// partitioner for `Producer`.
 ///
@@ -577,7 +581,7 @@ pub trait Partitioner {
 ///
 /// See `Builder::with_partitioner`.
 #[derive(Default)]
-pub struct DefaultPartitioner<H = BuildHasherDefault<SipHasher>> {
+pub struct DefaultPartitioner<H = BuildHasherDefault<DefaultHasher>> {
     // ~ a hasher builder; used to consistently hash keys
     hash_builder: H,
     // ~ a counter incremented with each partitioned message to
@@ -658,11 +662,11 @@ impl<H: BuildHasher> Partitioner for DefaultPartitioner<H> {
 
 #[cfg(test)]
 mod default_partitioner_tests {
-    use std::hash::{Hasher, SipHasher, BuildHasherDefault};
+    use std::hash::{Hasher, BuildHasherDefault};
     use std::collections::HashMap;
 
     use client;
-    use super::{DefaultPartitioner, Partitioner, Partitions, Topics};
+    use super::{DefaultPartitioner, DefaultHasher, Partitioner, Partitions, Topics};
 
     fn topics_map(topics: Vec<(&str, Partitions)>) -> HashMap<String, Partitions> {
         let mut h = HashMap::new();
@@ -703,7 +707,7 @@ mod default_partitioner_tests {
                                      num_all_partitions: 2,
                                  })]);
 
-        let mut p: DefaultPartitioner<BuildHasherDefault<SipHasher>> = Default::default();
+        let mut p: DefaultPartitioner<BuildHasherDefault<DefaultHasher>> = Default::default();
 
         // ~ validate that partitioning by the same key leads to the same
         // partition


Test patch: