Fix CustomHasher #841
Fix CustomHasher #841
Conversation
Passing an already instantiated hasher is a really bad idea. Instead pass a function returning the expected interface, so the hasher is instantiated when needed, also assuring there's one hasher for each partitionDispatcher and thus avoid concurrency problems."
partitioner.go
Outdated
| @@ -82,18 +82,20 @@ func (p *roundRobinPartitioner) RequiresConsistency() bool { | |||
| return false | |||
| } | |||
|
|
|||
| type hasherFunc func() hash.Hash32 | |||
There was a problem hiding this comment.
I wouldn't even bother naming this type, especially since it's weird for a non-exported interface to be an argument to an exported method.
partitioner.go
Outdated
| @@ -82,18 +82,20 @@ func (p *roundRobinPartitioner) RequiresConsistency() bool { | |||
| return false | |||
| } | |||
|
|
|||
| type hasherFunc func() hash.Hash32 | |||
|
|
|||
| type hashPartitioner struct { | |||
| random Partitioner | |||
| hasher hash.Hash32 | |||
| } | |||
|
|
|||
| // NewCustomHashPartitioner is a wrapper around NewHashPartitioner, | |||
| // allowing the use of custom hasher | |||
There was a problem hiding this comment.
might be worth documenting here that it expects the method reference not an instance of the hasher
There was a problem hiding this comment.
Absolutely. I might have gotten a bit in a hurry to open the PR, since I realised my mistake.
Remove type hasherFunc. Adjust doc comments for the NewCustomHashPartitioner.
|
Thanks! While technically this is a breaking API change the method was just introduced, broken in its current format, and never included in an actual release, so I think we can skirt the go rules for this one. |
|
Thank you too! Those were also my thoughts, plus the fact that without this change, it was simply broken. Also learned a lesson or two. |
I had to realise that even though now the same hashing algorithm was used across services to determine the partition, the problem that messages published under the same key on log compact topics still ended up on different partitions. Unfortunately I found out that I made a huge mistake by passing an already instantiated hasher. Thus the proposed fix: pass a callback so the hasher is instantiated just before needed (thanks to the wiki page for helping my understanding of the publisher's flow).