fix to prevent overlapping partitions in case of odd weight ratios (#98)

Author: laxmanchekka

kafka-streams-partitioners/weighted-group-partitioner/src/main/java/org/hypertrace/core/kafkastreams/framework/partitioner/WeightedGroupPartitioner.java

@@ -50,15 +50,20 @@ public WeightedGroupPartitioner(
   @Override
   public Integer partition(String topic, K key, V value, int numPartitions) {
     WeightedGroup groupConfig = this.getGroupConfig(topic, key, value);
-    int fromIndex = (int) Math.floor(groupConfig.getNormalizedFractionalStart() * numPartitions);
-    int toIndex = (int) Math.ceil(groupConfig.getNormalizedFractionalEnd() * numPartitions);
-    int numPartitionsForGroup = toIndex - fromIndex;
+    int fromIndexInclusive =
+        (int) Math.floor(groupConfig.getNormalizedFractionalStart() * numPartitions);
+    int toIndexExclusive =
+        (int) Math.floor(groupConfig.getNormalizedFractionalEnd() * numPartitions);
+    // Partition indexing starts from 0.
+    // Every group size should be at least one. This prevents divide by zero error in delegate
+    // partitioner.
+    int numPartitionsForGroup = Math.max(toIndexExclusive - fromIndexInclusive, 1);
 
     // partitioner by contract can return null.
     // Refer api doc:  org.apache.kafka.streams.processor.StreamPartitioner.partition
     // when delegate partitioner returns null, we use fallback partitioner (round-robin within
     // group)
-    return fromIndex
+    return fromIndexInclusive
         + Optional.ofNullable(
                 delegatePartitioner.partition(topic, key, value, numPartitionsForGroup))
             .orElse(

kafka-streams-partitioners/weighted-group-partitioner/src/test/java/org/hypertrace/core/kafkastreams/framework/partitioner/WeightedGroupPartitionerTest.java

@@ -175,13 +175,67 @@ public void testPartitionerWithKeyHashDelegatePartitioner() {
     assertEquals(4, partition);
   }
 
+  @Test
+  public void testWithNonMultipleWeightRatio() {
+
+    int testCount = 100;
+    PartitionerConfigServiceClient testClient =
+        (profileName) ->
+            new WeightedGroupProfile(
+                PartitionerProfile.newBuilder()
+                    .addGroups(newPartitionerGroup("group1", new String[] {"tenant-1"}, 27))
+                    .addGroups(
+                        newPartitionerGroup("group2", new String[] {"tenant-2", "tenant-3"}, 27))
+                    .setDefaultGroupWeight(52)
+                    .setName(profileName)
+                    .build());
+
+    WeightedGroupPartitioner<String, String> partitioner =
+        new WeightedGroupPartitioner<>(
+            "spans", testClient, groupKeyExtractor, roundRobinPartitioner);
+    int partition;
+
+    // Test case 1: tenant-1 belong to group-1 (partitions: [0 to 7])
+    for (int i = 1; i <= testCount; i++) {
+      partition = partitioner.partition("test-topic", "tenant-1", "span-" + i, 32);
+      assertTrue(
+          partition >= 0 && partition <= 7,
+          "actual partition not in expected range. partition: " + partition);
+    }
+
+    // Test case 2: tenant-2 belong to group-2 (partitions: [8 to 15])
+    for (int i = 1; i <= testCount; i++) {
+      partition = partitioner.partition("test-topic", "tenant-2", "span-" + i, 32);
+      assertTrue(
+          partition >= 8 && partition <= 15,
+          "actual partition not in expected range. partition: " + partition);
+    }
+
+    // Test case 3: tenant-3 belong to group-2 (partitions: [8 to 15])
+    for (int i = 1; i <= testCount; i++) {
+      partition = partitioner.partition("test-topic", "tenant-3", "span-" + i, 32);
+      assertTrue(
+          partition >= 8 && partition <= 15,
+          "actual partition not in expected range. partition: " + partition);
+    }
+
+    // Test case 4: groupKey=unknown should use default group [16 to 31]
+    for (int i = 1; i <= testCount; i++) {
+      partition = partitioner.partition("test-topic", "unknown", "span-" + i, 32);
+      assertTrue(
+          partition >= 16 && partition <= 31,
+          "actual partition not in expected range. partition: " + partition);
+    }
+  }
+
   private PartitionerConfigServiceClient getTestServiceClient() {
     return (profileName) ->
         new WeightedGroupProfile(
             PartitionerProfile.newBuilder()
                 .addGroups(newPartitionerGroup("group1", new String[] {"tenant-1"}, 25))
                 .addGroups(newPartitionerGroup("group2", new String[] {"tenant-2", "tenant-3"}, 25))
                 .setDefaultGroupWeight(50)
+                .setName(profileName)
                 .build());
   }