UladzislauBlok commented on code in PR #22570:
URL: https://github.com/apache/kafka/pull/22570#discussion_r3597487822
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/WindowedStreamPartitioner.java:
##########
@@ -33,21 +34,28 @@ public WindowedStreamPartitioner(final
WindowedSerializer<K> serializer) {
this.serializer = serializer;
}
+ @SuppressWarnings({"removal"})
+ @Override
+ public Optional<Set<Integer>> partitions(final String topic, final
Windowed<K> windowedKey, final V value, final int numPartitions) {
+ return partitions(topic, windowedKey, value, new RecordHeaders(),
numPartitions);
+ }
+
/**
* WindowedStreamPartitioner determines the partition number for a record
with the given windowed key and value
* and the current number of partitions. The partition number id
determined by the original key of the windowed key
* using the same logic as DefaultPartitioner so that the topic is
partitioned by the original key.
*
- * @param topic the topic name this record is sent to
- * @param windowedKey the key of the record
- * @param value the value of the record
+ * @param topic the topic name this record is sent to
+ * @param windowedKey the key of the record
+ * @param value the value of the record
+ * @param headers the record headers
* @param numPartitions the total number of partitions
* @return an integer between 0 and {@code numPartitions-1}, or {@code
null} if the default partitioning logic should be used
*/
@Override
- public Optional<Set<Integer>> partitions(final String topic, final
Windowed<K> windowedKey, final V value, final int numPartitions) {
+ public Optional<Set<Integer>> partitions(final String topic, final
Windowed<K> windowedKey, final V value, final Headers headers, final int
numPartitions) {
// for windowed key, the key bytes should never be null
- final byte[] keyBytes = serializer.serializeBaseKey(topic, new
RecordHeaders(), windowedKey);
+ final byte[] keyBytes = serializer.serializeBaseKey(topic, headers,
windowedKey);
Review Comment:
Checked code of `SessionWindowedSerializer` and `TimeWindowedSerializer`.
They both have inner serialized (the fact I missed)
I will update KIP (and include `DefaultStreamPartitioner` as well)
Thanks for catching this
--
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]
