DefaultPartitioner - implements java client logic

1 files changed, 23 insertions, 0 deletions

diff --git a/kafka/partitioner/default.py b/kafka/partitioner/default.py
new file mode 100644
index 0000000..358efeb
--- /dev/null
+++ b/kafka/partitioner/default.py
@@ -0,0 +1,23 @@
+import random
+
+from .hashed import murmur2
+
+
+class DefaultPartitioner(object):
+    """Default partitioner.
+
+    Hashes key to partition using murmur2 hashing (from java client)
+    If key is None, selects partition randomly from available,
+    or from all partitions if none are currently available
+    """
+    @classmethod
+    def __call__(cls, key, all_partitions, available):
+        if key is None:
+            if available:
+                return random.choice(available)
+            return random.choice(all_partitions)
+
+        idx = murmur2(key)
+        idx &= 0x7fffffff
+        idx %= len(all_partitions)
+        return all_partitions[idx]