Upgrade to Pro
— share decks privately, control downloads, hide ads and more …
Speaker Deck
Features
Speaker Deck
PRO
Sign in
Sign up for free
Search
Search
Kafka Partitioning Algorithm
Search
Gordon Diggs
September 02, 2016
Technology
0
140
Kafka Partitioning Algorithm
A look at the default partitioning algorithm that Kafka uses
Gordon Diggs
September 02, 2016
Tweet
Share
More Decks by Gordon Diggs
See All by Gordon Diggs
John Coltrane: Lessons in Leadership
gordondiggs
2
280
The Customer Gap
gordondiggs
1
110
Picking Records with JavaScript and a Button
gordondiggs
0
77
Supbutton
gordondiggs
0
61
Rayons
gordondiggs
0
63
Sous Vide
gordondiggs
0
81
Dev Events & Internal Tools at Paperless Post
gordondiggs
0
120
The Joys and Pains of Working With an Old Codebase
gordondiggs
0
130
The Joys and Pains of Working with an Old Codebase
gordondiggs
1
2.3k
Other Decks in Technology
See All in Technology
pprof vs runtime/trace (FlightRecorder)
task4233
0
170
バイブコーディングと継続的デプロイメント
nwiizo
2
470
多野優介
tanoyusuke
1
480
リーダーになったら未来を語れるようになろう/Speak the Future
sanogemaru
0
320
GC25 Recap+: Advancing Go Garbage Collection with Green Tea
logica0419
1
430
GA technologiesでのAI-Readyの取り組み@DataOps Night
yuto16
0
280
about #74462 go/token#FileSet
tomtwinkle
1
460
カンファレンスに託児サポートがあるということ / Having Childcare Support at Conferences
nobu09
1
260
Azure SynapseからAzure Databricksへ 移行してわかった新時代のコスト問題!?
databricksjapan
0
150
研究開発部メンバーの働き⽅ / Sansan R&D Profile
sansan33
PRO
3
20k
JAZUG 15周年記念 × JAT「AI Agent開発者必見:"今"のOracle技術で拡張するAzure × OCIの共存アーキテクチャ」
shisyu_gaku
0
130
LLM時代にデータエンジニアの役割はどう変わるか?
ikkimiyazaki
4
1k
Featured
See All Featured
Why Our Code Smells
bkeepers
PRO
339
57k
4 Signs Your Business is Dying
shpigford
185
22k
Facilitating Awesome Meetings
lara
56
6.6k
Art, The Web, and Tiny UX
lynnandtonic
303
21k
Refactoring Trust on Your Teams (GOTO; Chicago 2020)
rmw
35
3.2k
Let's Do A Bunch of Simple Stuff to Make Websites Faster
chriscoyier
507
140k
[RailsConf 2023] Rails as a piece of cake
palkan
57
5.9k
ReactJS: Keep Simple. Everything can be a component!
pedronauck
667
120k
RailsConf & Balkan Ruby 2019: The Past, Present, and Future of Rails at GitHub
eileencodes
140
34k
Exploring the Power of Turbo Streams & Action Cable | RailsConf2023
kevinliebholz
34
6.1k
The Psychology of Web Performance [Beyond Tellerrand 2023]
tammyeverts
49
3.1k
The Straight Up "How To Draw Better" Workshop
denniskardys
237
140k
Transcript
Kafka Partitioning Algorithm CC LnL 20160902
Kafka Partitioning Algorithm CC LnL 20160902
Kafka Partitioning • Can be keyed • Round robin (mostly)
by default
public int partition(String topic, Object key, byte[] keyBytes, Object value,
byte[] valueBytes, Cluster cluster) { List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); int numPartitions = partitions.size(); if (keyBytes == null) { int nextValue = counter.getAndIncrement(); List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic); if (availablePartitions.size() > 0) { int part = Utils.toPositive(nextValue) % availablePartitions.size(); return availablePartitions.get(part).partition(); } else { // no partitions are available, give a non-available partition return Utils.toPositive(nextValue) % numPartitions; } } else { // hash the keyBytes to choose a partition return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; } }
public int partition(String topic, Object key, byte[] keyBytes, Object value,
byte[] valueBytes, Cluster cluster) { List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); int numPartitions = partitions.size(); if (keyBytes == null) { int nextValue = counter.getAndIncrement(); List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic); if (availablePartitions.size() > 0) { int part = Utils.toPositive(nextValue) % availablePartitions.size(); return availablePartitions.get(part).partition(); } else { // no partitions are available, give a non-available partition return Utils.toPositive(nextValue) % numPartitions; } } else { // hash the keyBytes to choose a partition return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; } }
public int partition(String topic, Object key, byte[] keyBytes, Object value,
byte[] valueBytes, Cluster cluster) { List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); int numPartitions = partitions.size(); if (keyBytes == null) { int nextValue = counter.getAndIncrement(); List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic); if (availablePartitions.size() > 0) { int part = Utils.toPositive(nextValue) % availablePartitions.size(); return availablePartitions.get(part).partition(); } else { // no partitions are available, give a non-available partition return Utils.toPositive(nextValue) % numPartitions; } } else { // hash the keyBytes to choose a partition return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; } }
public int partition(String topic, Object key, byte[] keyBytes, Object value,
byte[] valueBytes, Cluster cluster) { List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); int numPartitions = partitions.size(); if (keyBytes == null) { int nextValue = counter.getAndIncrement(); List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic); if (availablePartitions.size() > 0) { int part = Utils.toPositive(nextValue) % availablePartitions.size(); return availablePartitions.get(part).partition(); } else { // no partitions are available, give a non-available partition return Utils.toPositive(nextValue) % numPartitions; } } else { // hash the keyBytes to choose a partition return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; } }
public int partition(String topic, Object key, byte[] keyBytes, Object value,
byte[] valueBytes, Cluster cluster) { List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); int numPartitions = partitions.size(); if (keyBytes == null) { int nextValue = counter.getAndIncrement(); List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic); if (availablePartitions.size() > 0) { int part = Utils.toPositive(nextValue) % availablePartitions.size(); return availablePartitions.get(part).partition(); } else { // no partitions are available, give a non-available partition return Utils.toPositive(nextValue) % numPartitions; } } else { // hash the keyBytes to choose a partition return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; } }
public int partition(String topic, Object key, byte[] keyBytes, Object value,
byte[] valueBytes, Cluster cluster) { List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); int numPartitions = partitions.size(); if (keyBytes == null) { int nextValue = counter.getAndIncrement(); List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic); if (availablePartitions.size() > 0) { int part = Utils.toPositive(nextValue) % availablePartitions.size(); return availablePartitions.get(part).partition(); } else { // no partitions are available, give a non-available partition return Utils.toPositive(nextValue) % numPartitions; } } else { // hash the keyBytes to choose a partition return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; } }
public int partition(String topic, Object key, byte[] keyBytes, Object value,
byte[] valueBytes, Cluster cluster) { List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); int numPartitions = partitions.size(); if (keyBytes == null) { int nextValue = counter.getAndIncrement(); List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic); if (availablePartitions.size() > 0) { int part = Utils.toPositive(nextValue) % availablePartitions.size(); return availablePartitions.get(part).partition(); } else { // no partitions are available, give a non-available partition return Utils.toPositive(nextValue) % numPartitions; } } else { // hash the keyBytes to choose a partition return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; } }
public int partition(String topic, Object key, byte[] keyBytes, Object value,
byte[] valueBytes, Cluster cluster) { List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); int numPartitions = partitions.size(); if (keyBytes == null) { int nextValue = counter.getAndIncrement(); List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic); if (availablePartitions.size() > 0) { int part = Utils.toPositive(nextValue) % availablePartitions.size(); return availablePartitions.get(part).partition(); } else { // no partitions are available, give a non-available partition return Utils.toPositive(nextValue) % numPartitions; } } else { // hash the keyBytes to choose a partition return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; } }
Utils.toPositive
public static int toPositive(int number) { return number & 0x7fffffff;
}
0111 1111 1111 1111 1111 1111 1111 1111
0000 0000 0000 0000 0000 0000 0000 1010 & 0111
1111 1111 1111 1111 1111 1111 1111 = 0000 0000 0000 0000 0000 0000 0000 1010
Two’s Complement
0000 0000 0000 0000 0000 0000 0000 1010 1111 1111
1111 1111 1111 1111 1111 0101 1111 1111 1111 1111 1111 1111 1111 1010
1111 1111 1111 1111 1111 1111 1111 1010 & 0111
1111 1111 1111 1111 1111 1111 1111 = 0111 1111 1111 1111 1111 1111 1111 1010
None