write_kafka: ensure key is properly hashed to select partition

Otherwise, partition is selected solely on the first letter of the
key. If all your hosts are starting with the same letter, everything
ends up in a single partition.

diff --git a/src/write_kafka.c b/src/write_kafka.c
index e881593b65161e20af6b697dd122ca2e402829c0..614ce0f084df7f9ba09181b21983a2fdbec277a5 100644 (file)
--- a/src/write_kafka.c
+++ b/src/write_kafka.c
@@ -73,11 +73,19 @@ static void kafka_log(const rd_kafka_t *rkt, int level,
 }
 #endif
 
+static uint32_t kafka_hash(const char *keydata, size_t keylen)
+{
+    uint32_t hash = 5381;
+    for (; keylen > 0; keylen--)
+        hash = ((hash << 5) + hash) + keydata[keylen - 1];
+    return hash;
+}
+
 static int32_t kafka_partition(const rd_kafka_topic_t *rkt,
                                const void *keydata, size_t keylen,
                                int32_t partition_cnt, void *p, void *m)
 {
-    uint32_t key = *((uint32_t *)keydata );
+    uint32_t key = kafka_hash(keydata, keylen);
     uint32_t target = key % partition_cnt;
     int32_t  i = partition_cnt;