aliehsaeedii commented on code in PR #22570:
URL: https://github.com/apache/kafka/pull/22570#discussion_r3532075182
##########
streams/src/main/java/org/apache/kafka/streams/processor/StreamPartitioner.java:
##########
@@ -53,17 +54,23 @@
@FunctionalInterface
public interface StreamPartitioner<K, V> {
+ @Deprecated(since = "4.4", forRemoval = true)
+ Optional<Set<Integer>> partitions(String topic, K key, V value, int
numPartitions);
Review Comment:
The deprecated method lost its javadoc entirely (it moved to the new
overload). Since this stays part of the public API until removal, it should
keep a `@deprecated` javadoc tag pointing at the replacement, e.g.:
```java
/**
* @deprecated Since 4.4. Use {@link #partitions(String, Object, Object,
Headers, int)} instead.
*/
```
This also answers the question from the other thread: yes, the javadoc tag
is expected alongside the annotation.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/KafkaStreamsNamedTopologyWrapper.java:
##########
@@ -416,15 +417,15 @@ public Collection<StreamsMetadata>
allStreamsClientsMetadataForTopology(final St
}
/**
- * See {@link KafkaStreams#queryMetadataForKey(String, Object, Serializer)}
+ * See {@link KafkaStreams#queryMetadataForKey(String, Object, Headers,
Serializer)}
Review Comment:
`Headers` is referenced in this `{@link}` but only
`org.apache.kafka.common.header.internals.RecordHeaders` was imported, so the
javadoc reference won't resolve. Please import
`org.apache.kafka.common.header.Headers` as well (or fully qualify it in the
link).
##########
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:
Behavioral note worth confirming against the KIP: previously the base key
was always serialized with empty `RecordHeaders`; now it gets the real record
headers. For applications whose windowed key serializer is header-sensitive,
the partition assignment of windowed sink/repartition records silently changes
after upgrading, which can break co-partitioning with data written before the
upgrade. If the KIP already covers this, it should at least be called out in
the upgrade notes / streams docs PR you mentioned.
##########
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java:
##########
@@ -1735,38 +1737,66 @@ public Collection<StreamsMetadata>
streamsMetadataForStore(final String storeNam
return streamsMetadataState.allMetadataForStore(storeName);
}
+ /**
+ * Finds the metadata containing the active hosts and standby hosts where
the key being queried would reside,
+ * without requiring record headers to be provided.
+ * <p>
+ * See {@link #queryMetadataForKey(String, Object, Headers, Serializer)}
for more details.
+ */
+ public <K> KeyQueryMetadata queryMetadataForKey(final String storeName,
+ final K key,
+ final Serializer<K>
keySerializer) {
+ return queryMetadataForKey(storeName, key, new RecordHeaders(),
keySerializer);
Review Comment:
For users whose partitioner (or key serializer) is header-aware, this
convenience overload computes the partition from *empty* headers, so it can
locate the wrong host for keys that were written with different headers. Worth
a sentence in the javadoc of both no-headers overloads: "if your partitioner or
serializer makes use of headers, use the `Headers` overload, otherwise the
returned metadata may not match where the key actually resides."
##########
streams/src/main/java/org/apache/kafka/streams/processor/StreamPartitioner.java:
##########
@@ -53,17 +54,23 @@
@FunctionalInterface
public interface StreamPartitioner<K, V> {
+ @Deprecated(since = "4.3", forRemoval = true)
+ Optional<Set<Integer>> partitions(String topic, K key, V value, int
numPartitions);
+
/**
- * Determine the number(s) of the partition(s) to which a record with the
given key and value should be sent,
+ * Determine the number(s) of the partition(s) to which a record with the
given key and value should be sent,
* for the given topic and current partition count
* @param topic the topic name this record is sent to
* @param key 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 Optional of Set of integers between 0 and {@code
numPartitions-1},
* Empty optional means using default partitioner
* Optional of an empty set means the record won't be sent to any
partitions i.e drop it.
* Optional of Set of integers means the partitions to which the record
should be sent to.
* */
- Optional<Set<Integer>> partitions(String topic, K key, V value, int
numPartitions);
+ default Optional<Set<Integer>> partitions(final String topic, final K key,
final V value, final Headers headers, final int numPartitions) {
Review Comment:
Thanks — agreed. Fine to keep as is. +1 on documenting the lambda limitation
in the streams docs in a separate PR — good that it is tracked on the ticket:)
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java:
##########
@@ -1074,7 +1074,13 @@ private void assertNextOutputRecord(final
TestRecord<String, String> record,
}
private StreamPartitioner<String, String> constantPartitioner(final
Integer partition) {
- return (topic, key, value, numPartitions) ->
Optional.of(Collections.singleton(partition));
+ return new StreamPartitioner<>() {
+ @SuppressWarnings("removal")
+ @Override
+ public Optional<Set<Integer>> partitions(final String topic, final
String key, final String value, final int numPartitions) {
+ return Optional.of(Collections.singleton(partition));
+ }
+ };
}
Review Comment:
Since the KIP's core compatibility promise is exactly that existing lambda
implementations keep working, should we consider keeping at least one
lambda-based `StreamPartitioner` in the tests so that promise stays guarded at
compile time and runtime. (Converting to an anonymous class is of course right
for the tests that assert the deprecated method is NOT called.)
##########
streams/src/main/java/org/apache/kafka/streams/processor/StreamPartitioner.java:
##########
@@ -53,17 +54,23 @@
@FunctionalInterface
public interface StreamPartitioner<K, V> {
+ @Deprecated(since = "4.3", forRemoval = true)
Review Comment:
yes to the javadoc: after moving the docs to the new overload this method
has no javadoc at all, and the generated API docs expect a `@deprecated` tag
alongside the annotation. Something like: `@deprecated Since 4.4. Use {@link
#partitions(String, Object, Object, Headers, int)} instead.` Re mentioning
removal: "will be removed in a future major release".
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/WindowedStreamPartitionerTest.java:
##########
@@ -81,4 +82,29 @@ public void testCopartitioning() {
}
}
}
+
+ @Test
+ public void testCopartitioningWithHeaders() {
+ final Random rand = new Random();
+ final org.apache.kafka.common.header.Headers headers = new
org.apache.kafka.common.header.internals.RecordHeaders();
Review Comment:
nit: use imports for `Headers`/`RecordHeaders` instead of fully-qualified
names inline — consistent with the rest of the codebase (and with
`DefaultStreamPartitionerTest` in this same PR).
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java:
##########
@@ -239,13 +244,15 @@ public synchronized <K> KeyQueryMetadata
keyQueryMetadataForKey(final String sto
*
* @param storeName Name of the store
* @param key Key to use
+ * @param headers the record headers
* @param partitioner partitioner to use to find correct partition for key
* @param <K> key type
* @return The {@link KeyQueryMetadata} for the storeName and key or
{@link KeyQueryMetadata#NOT_AVAILABLE}
* if streams is (re-)initializing, or {@code null} if no matching
metadata could be found.
*/
public synchronized <K> KeyQueryMetadata keyQueryMetadataForKey(final
String storeName,
final K
key,
Review Comment:
`storeName`, `key`, and `partitioner`/`keySerializer` are all null-checked
here, but the new `headers` parameter is not. A caller passing `null` via
`KafkaStreams#queryMetadataForKey(store, key, null, serializer)` would flow
`null` into header-aware partitioners/serializers and NPE deep inside user
code. Consider either `Objects.requireNonNull(headers, "headers can't be
null")` or normalizing `null` to `new RecordHeaders()` for consistency with the
no-headers overloads.
--
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]
