junaiddshaukat commented on code in PR #39610:
URL: https://github.com/apache/beam/pull/39610#discussion_r3711238092
##########
runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/GroupByKeyBroadcastPartitioner.java:
##########
@@ -56,7 +56,10 @@ public Optional<Set<Integer>> partitions(
}
return Optional.of(all);
}
- int partition = Utils.toPositive(Utils.murmur2(key)) % numPartitions;
+ // A keyless record — a stateless stage carries no key — has nowhere in
particular to go, so
+ // send it to partition 0 rather than hashing a null. This is the method
Kafka Streams calls,
+ // so the guard has to be here and not only on partition() above.
+ int partition = key == null ? 0 : Utils.toPositive(Utils.murmur2(key)) %
numPartitions;
return Optional.of(Collections.singleton(partition));
Review Comment:
You're right I checked again, partition 0 is a fixed partition, not an
absence of one — keyless records would all have piled onto it. Fixed by
returning Optional.empty() instead: I checked RecordCollectorImpl, and it tests
isPresent() before using the returned set, so an empty Optional means no
explicit partition was chosen and the producer's default partitioner handles
it. That spreads keyless records over the topic, which is the behaviour a
record with no key should get.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
