Make the default partitioner topic-aware when in round-robin mode

src/producer/partitioners/default/index.spec.js

@@ -51,6 +51,30 @@ describe('Producer > Partitioner > Default', () => {
     expect(partitionCount[2]).toEqual(10)
   })
 
+  test('messages are partitioned in a round-robin fashion for each topic', () => {
+    const partitionMetadata = [
+      { partitionId: 1, leader: 1 },
+      { partitionId: 2, leader: 2 },
+    ]
+    const topics = [
+      { topic: 'topic-a', partitionMetadata, partitionCount: {} },
+      { topic: 'topic-b', partitionMetadata, partitionCount: {} },
+    ]
+
+    for (let i = 0; i < 30; ++i) {
+      for (const { topic, partitionMetadata, partitionCount } of topics) {
+        const partition = partitioner({ topic, partitionMetadata, message: {} })
+        const count = partitionCount[partition] || 0
+        partitionCount[partition] = count + 1
+      }
+    }
+
+    expect(topics[0].partitionCount[1]).toEqual(15)
+    expect(topics[0].partitionCount[2]).toEqual(15)
+    expect(topics[1].partitionCount[1]).toEqual(15)
+    expect(topics[1].partitionCount[2]).toEqual(15)
+  })
+
   test('returns the configured partition if it exists', () => {
     const partition = partitioner({
       topic,

src/producer/partitioners/default/partitioner.js

@@ -18,9 +18,12 @@ const toPositive = x => x & 0x7fffffff
  *  - If no partition or key is present choose a partition in a round-robin fashion
  */
 module.exports = murmur2 => () => {
-  let counter = randomBytes(32).readUInt32BE(0)
+  const counters = {}
 
   return ({ topic, partitionMetadata, message }) => {
+    if (!(topic in counters)) {
+      counters[topic] = randomBytes(32).readUInt32BE(0)
+    }
     const numPartitions = partitionMetadata.length
     const availablePartitions = partitionMetadata.filter(p => p.leader >= 0)
     const numAvailablePartitions = availablePartitions.length
@@ -34,11 +37,11 @@ module.exports = murmur2 => () => {
     }
 
     if (numAvailablePartitions > 0) {
-      const i = toPositive(++counter) % numAvailablePartitions
+      const i = toPositive(++counters[topic]) % numAvailablePartitions
       return availablePartitions[i].partitionId
     }
 
     // no partitions are available, give a non-available partition
-    return toPositive(++counter) % numPartitions
+    return toPositive(++counters[topic]) % numPartitions
   }
 }