X-Git-Url: https://git.verplant.org/?a=blobdiff_plain;f=src%2Fwrite_kafka.c;h=614ce0f084df7f9ba09181b21983a2fdbec277a5;hb=ea9db2c32ef7f8a3135dc1413fbf8ee46a25f680;hp=e881593b65161e20af6b697dd122ca2e402829c0;hpb=5eb1fef6de7e0c8fb99e3d85063d96d7cc785f9b;p=collectd.git diff --git a/src/write_kafka.c b/src/write_kafka.c index e881593b..614ce0f0 100644 --- 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;