Murmur2 partitioner #1468

barrotsteindev · 2017-10-18T19:59:25Z

Add murmur2 partitioner, currently using the fastest algorithm for murmur2, which is not endian neutral. Also implements an endian neutral algorithm which can be used instead.
Fixes part of issue #1424

edenhill · 2017-10-19T10:59:11Z

Thanks for this, much appreciated and anticipated!

I will review this more careful in a while, but from a quick glance there are some low hanging fruit that needs fixing:

remove all white space diffs
remove the CONFIGURATION.md changes (it is auto generated from your build options)
add a LICENSE.murmur2 file to the top level
is there any proof of correctness for the provided implementation?
I'd like to see some unit tests to verify it produces the same result as the Java counterpart (look at rd_varint.c and rd_unittest.c to see how unit tests are added)
I dont think the new rdkafka_simple_producer_murmur2 example is warranted, about 99% of it is duplicated from existing examples. It might be better to include it in one of the other examples as an option.
There are comments in the murmur code that it must be able to perform unaligned reads, this is not portable.
There are also endian comments in the murmur code. Does it work on big-endian?
Please follow the coding guideline: use 8 whitespaces for tab

barrotsteindev · 2017-10-19T19:24:36Z

I will use the endian neutral hash.
The murmur2hash.c file is from a repo that has no direct mention of a license.
Will fix the tabs and configuration file.
Thanks for your patience I'm fairly new to this.

barrotsteindev · 2017-10-20T06:14:44Z

this should probably be the hash function (MurmurHashNeutral2) which will be used by librdkafka,
since it seems to be endian neutral and can aligned readable

// MurmurHashNeutral2, by Austin Appleby
//
// Same as MurmurHash2, but endian- and alignment-neutral.
// Half the speed though, alas.
*/

would be lovely if another set of eyes would take a look at that implementation to see if there's anything suspicious. I will test it against the java implementation shortly

barrotsteindev · 2017-10-22T16:20:25Z

I hope I addressed all of your requests with these commits

edenhill

Looking better.
Still a few issues to sort out, and some massive whitespace diffs that needs to go away.
Also make sure to reading the coding guidelines regarding formatting and indentation.

edenhill · 2017-10-23T12:08:22Z

LICENSE.murmur2

+// MurmurHash2 was written by Austin Appleby, and is placed in the public
+// domain. The author hereby disclaims copyright to this source code.
+
+Permission is hereby granted, free of charge, to any person obtaining a copy


Does the following license really apply to public domain?
Did Austin Appleby specify this license for the code?

yes AppleBy distributed his software under the MIT license.

edenhill · 2017-10-23T12:08:39Z

examples/Makefile

@@ -89,4 +89,3 @@ rdkafka_zookeeper_example: ../src/librdkafka.a rdkafka_zookeeper_example.c

 clean:
 	rm -f $(EXAMPLES)
-


remove whitespace diff

edenhill · 2017-10-23T12:11:14Z

examples/rdkafka_simple_producer.c

-        if (argc != 3) {
-                fprintf(stderr, "%% Usage: %s <broker> <topic>\n", argv[0]);
-                return 1;
+        const char *brokers = NULL;    /* Argument: broker list */


The rdkafka_simple_producer was added because rdkafka_example.c was getting heavy with options trying to show-case everything.
I'd like the simple producer to stay simple.

We can wait with example code until the config-string-based partitioner support is in place (so you could do conf_set("partitioner", "murmur2", ..)).

So please remove the patch for this file.

edenhill · 2017-10-23T12:14:31Z

src/MurMurHash2.h

+    extern "C" {
+  #endif
+
+  void SetAffinity ( int cpu );


This doesn't seem to be used anywhere, remove.

edenhill · 2017-10-23T12:15:26Z

src/rdcrc32.h

@@ -30,10 +30,6 @@
 #include <zlib.h>
 #endif

-#ifdef __cplusplus


This should not be removed.

edenhill · 2017-10-23T12:19:59Z

src/rdmurmur2.c

+ #include "rdmurmur2.h"
+
+int unittest_murmurhashneutral2 (void) {
+  const char * keysToTest[] = {


indent: 8 whitespaces.

Format: const char *keysToTest[] .. (pay attention to * placement)

edenhill · 2017-10-23T12:21:35Z

src/rdmurmur2.c

+  const char * keysToTest[] = {
+    "kafka",
+    "amqp",
+    "giberish123456789"


Since the murmur2 code is prone to alignment and word issues I suggest you also add the following tests:

unaligned access

word-boundaries (i.e., lengths 7,8,9, 15,16,17)

i have fixed all issues but this. What did you mean by word boundaries, so i allocate pointers and start at an address offset by 7 bits from the aligned address?
this is the only thing blocking me right now

edenhill · 2017-10-23T12:23:07Z

src/rdmurmur2.h

+#ifndef __RDMURMUR2___H__
+#define __RDMURMUR2___H__
+
+#ifdef __cplusplus


not needed in C code, remove.

edenhill · 2017-10-23T12:23:41Z

src/rdmurmur2.h

+
+static RD_INLINE rd_murmur2_t rd_murmur2(const char *data, size_t data_len) {
+  /*
+   * last bit is set to 0 because the java implementation uses int_32


Isn't a cast to int32_t sufficient?

no since the java version flips the first bit to make it positive and we need to maintain that behaviour

edenhill · 2017-10-23T12:24:28Z

win32/librdkafka.vcxproj

@@ -155,6 +156,7 @@
  </ItemGroup>
  <ItemGroup>
    <ClCompile Include="..\src\crc32c.c" />
+    <ClCompile Include="..\src\MurMurHash2.c" />
    <ClCompile Include="..\src\rdaddr.c" />


what about rdmurmur2.h?

edenhill

Can you also please rebase this on current master and fixup / squash the commits?

edenhill · 2017-10-28T07:57:11Z

src/rdkafka_msg.c

+					 int32_t partition_cnt,
+					 void *rkt_opaque,
+					 void *msg_opaque) {
+	if (keylen == 0)


Looking at the Java partitioner I believe this should probably be if (!key) since there is a difference between no key (null) and an empty key (which is just an arbitrary key).
If so this is also incorrect on the existing partitioners, but we can't change their behaviour.

barrotsteindev · 2017-10-28T12:53:06Z

Sure thing, I will change it.

…

On Oct 28, 2017 10:59 AM, "Magnus Edenhill" ***@***.***> wrote: ***@***.**** requested changes on this pull request. Can you also please rebase this on current master and fixup / squash the commits? ------------------------------ In src/rdkafka_msg.c <#1468 (comment)>: > @@ -570,6 +541,35 @@ int32_t rd_kafka_msg_partitioner_consistent_random (const rd_kafka_topic_t *rkt, msg_opaque); } +int32_t rd_kafka_msg_partitioner_murmur2_consistent (const rd_kafka_topic_t *rkt, + const void *key, size_t keylen, + int32_t partition_cnt, + void *rkt_opaque, + void *msg_opaque) { + return rd_murmur2(key, keylen) % partition_cnt; +} + +int32_t rd_kafka_msg_partitioner_murmur2_random (const rd_kafka_topic_t *rkt, + const void *key, size_t keylen, + int32_t partition_cnt, + void *rkt_opaque, + void *msg_opaque) { + if (keylen == 0) Looking at the Java partitioner I believe this should probably be if (!key) since there is a difference between no key (null) and an empty key (which is just an arbitrary key). If so this is also incorrect on the existing partitioners, but we can't change their behaviour. — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#1468 (review)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/ACWANHTtwFhXS54IjnMNoW6ohCdWvBHIks5swt7lgaJpZM4P-Rip> .

barrotsteindev · 2017-11-04T17:58:03Z

@edenhill I have implemented most fixes except for the word boundary. Could you give a hint because I did not quite understand the unit test you requested. I would really appreciate it.

tmichaud314 · 2017-12-08T14:09:57Z

@barrotsteindev I would love to see this PR happen and it looks close to completion. Do you foresee this getting merged? Thanks for your efforts here!

barrotsteindev · 2017-12-10T19:22:15Z

i have rebased this pull request and responded to your requests @edenhill

edenhill

Looks good, but the whitespace diffs needs to be removed and the other formatting comments need to be addressed as well.

edenhill · 2017-12-11T07:00:11Z

src/rdkafka_msg.c

@@ -3,24 +3,24 @@
 *
 * Copyright (c) 2012,2013 Magnus Edenhill
 * All rights reserved.
- * 
+ *


remove whitespace diffs

edenhill · 2017-12-11T07:00:21Z

examples/Makefile

@@ -89,4 +89,3 @@ rdkafka_zookeeper_example: ../src/librdkafka.a rdkafka_zookeeper_example.c

 clean:
 	rm -f $(EXAMPLES)
-


remove whitespace diff

edenhill · 2017-12-11T07:00:34Z

src/rdkafka_msg.c

@@ -222,7 +223,7 @@ int rd_kafka_msg_new (rd_kafka_itopic_t *rkt, int32_t force_partition,
 	int errnox;

        /* Create message */
-        rkm = rd_kafka_msg_new0(rkt, force_partition, msgflags, 


remove whitespace diff

edenhill · 2017-12-11T07:00:59Z

src/rdkafka_msg.c

+					 void *msg_opaque) {
+	if (!key)
+		return rd_kafka_msg_partitioner_random(rkt,
+																					 key,


remove empty lines

edenhill · 2017-12-11T07:01:32Z

src/rdkafka_msg.c

@@ -799,4 +829,3 @@ int64_t rd_kafka_message_latency (const rd_kafka_message_t *rkmessage) {

        return rd_clock() - rkm->rkm_ts_enq;
 }
-


remove whitespace diff

Whitespace diff line 225

Fixed typo

barrotsteindev · 2017-12-12T08:27:36Z

Hey @edenhill,
I addressed your new review, hope it's satisfactory.
Thanks in advance.

edenhill · 2017-12-12T08:34:29Z

Looks good, thanks!
Just need to fix these OSX compile warnings and then we're good to merge:
https://travis-ci.org/edenhill/librdkafka/jobs/315112979#L229

barrotsteindev · 2017-12-12T18:04:26Z

@edenhill , I have examined the warnings from the osx build,
and have noticed that they only come from the non endian neutral hash functions, and the aligned only hash function. These functions are not used.
We can either ignore those warnings, since we do not use that code, or I can remove it from the file.
Which action do you prefer?
thanks in advance.

edenhill · 2017-12-12T20:00:12Z

We should probably use the optimized versions if the platform supports it, but we can fix that later.

edenhill · 2017-12-12T20:01:41Z

Big thanks for all the effort on this!

edenhill suggested changes Oct 23, 2017

View reviewed changes

edenhill suggested changes Oct 28, 2017

View reviewed changes

vuza mentioned this pull request Oct 31, 2017

Implementing Murmur2 partitioner, matching the Java client's partitioning logic nodefluent/node-sinek#33

Closed

barrotsteindev and others added 20 commits December 10, 2017 21:01

set_partitioner

19e070b

addded MurMur2

cdfce39

working MurMur2

f8fdb82

murmur2_consistent and random

ca437c4

move murmur2 to rdmurmur2.h

3bed904

added copyrights

c229a03

changed murmur2 example name

3572cbb

remove uneede strings.h

d8f890f

typo

839065a

Fix rdmumur2.h for cpp

803e35b

fix typo

2dbdc88

use endian neutral murmur2hash

45bec76

updated licenses

82dc199

added unitests for murmur2 hash

a2e7bf6

uppercase murmur2 unittest output

535629c

use merge murmur2 into example

108b3ff

use configuration.md from master branch

a6c3172

added MurMurHash2 deps

5d8d049

added src/rdmurmur2.c to vs proj

6ef1462

declare counter outside for loop

d37c1ce

Bar Rotstein added 3 commits December 10, 2017 21:01

removed unused func SetAffinity

b4c14d2

rdcrc32.h as the master branch

923f2c6

rever src/rdkafka.h changes

bcf7d84

barrotsteindev force-pushed the murmur2_partitioner branch from c51e2fc to bcf7d84 Compare December 10, 2017 19:12

Bar Rotstein added 3 commits December 10, 2017 21:19

!key -> review change

3359045

revert example changes

10a734f

!key -> review change

719768d

restructuring accroding to review

43c9d6a

edenhill suggested changes Dec 11, 2017

View reviewed changes

Bar Rotstein and others added 6 commits December 11, 2017 19:28

indent changes

ee26a5b

structuring

cec2fde

msg.c reformat

2bcd8ad

copy msg copyright

c986419

Update rdkafka_msg.c

a051b57

Whitespace diff line 225

Update rdmurmur2.c

83e212d

Fixed typo

edenhill merged commit 35f28e2 into confluentinc:master Dec 12, 2017

mhowlett mentioned this pull request Dec 13, 2017

How to use a custom partitioner ? confluentinc/confluent-kafka-dotnet#343

Closed

		@@ -89,4 +89,3 @@ rdkafka_zookeeper_example: ../src/librdkafka.a rdkafka_zookeeper_example.c

		clean:
		rm -f $(EXAMPLES)

		@@ -799,4 +829,3 @@ int64_t rd_kafka_message_latency (const rd_kafka_message_t *rkmessage) {

		return rd_clock() - rkm->rkm_ts_enq;
		}

Murmur2 partitioner #1468

Murmur2 partitioner #1468

Conversation

barrotsteindev commented Oct 18, 2017

edenhill commented Oct 19, 2017

barrotsteindev commented Oct 19, 2017

barrotsteindev commented Oct 20, 2017

barrotsteindev commented Oct 22, 2017

edenhill left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

edenhill left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

barrotsteindev commented Oct 28, 2017 via email

barrotsteindev commented Nov 4, 2017

tmichaud314 commented Dec 8, 2017

barrotsteindev commented Dec 10, 2017

edenhill left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

barrotsteindev commented Dec 12, 2017

edenhill commented Dec 12, 2017

barrotsteindev commented Dec 12, 2017

edenhill commented Dec 12, 2017

edenhill commented Dec 12, 2017