mjsax commented on code in PR #22570:
URL: https://github.com/apache/kafka/pull/22570#discussion_r3592266977
##########
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java:
##########
@@ -1780,38 +1782,70 @@ 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>
+ * 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.
Review Comment:
```suggestion
* If your partitioner or serializer makes use of headers, use the
{@link #queryMetadataForKey(String, Object, Headers, Serializer) Headers}
overload, otherwise the returned metadata may not match where the key actually
resides.
```
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java:
##########
@@ -1438,4 +1436,78 @@ private static void maybeSetOutputVersioned(final
GraphNode tableNode,
tableNode.setOutputVersioned(materializedInternal.storeSupplier()
instanceof VersionedBytesStoreSupplier);
}
}
+
+ private static class SubscriptionSinkPartitioner<K, KO> implements
StreamPartitioner<KO, SubscriptionWrapper<K>> {
+ private final StreamPartitioner<KO, Void> otherPartitioner;
+ private final Function<Optional<Set<Integer>>, Optional<Set<Integer>>>
getPartition;
+
+ private SubscriptionSinkPartitioner(
+ final StreamPartitioner<KO, Void> otherPartitioner,
+ final Function<Optional<Set<Integer>>, Optional<Set<Integer>>>
getPartition
+ ) {
+ this.otherPartitioner = otherPartitioner;
+ this.getPartition = getPartition;
+ }
+
+ @SuppressWarnings("removal")
+ @Override
+ public Optional<Set<Integer>> partitions(final String topic, final KO
key, final SubscriptionWrapper<K> value, final int numPartitions) {
+ return partitions(topic, key, value, new RecordHeaders(),
numPartitions);
Review Comment:
Given that this is all internal code, and we know that KS should never call
the deprecated overload, I am wondering if we should not forward the call here,
but throw an exception to catch any potential bugs?
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java:
##########
@@ -1438,4 +1436,78 @@ private static void maybeSetOutputVersioned(final
GraphNode tableNode,
tableNode.setOutputVersioned(materializedInternal.storeSupplier()
instanceof VersionedBytesStoreSupplier);
}
}
+
+ private static class SubscriptionSinkPartitioner<K, KO> implements
StreamPartitioner<KO, SubscriptionWrapper<K>> {
+ private final StreamPartitioner<KO, Void> otherPartitioner;
+ private final Function<Optional<Set<Integer>>, Optional<Set<Integer>>>
getPartition;
+
+ private SubscriptionSinkPartitioner(
+ final StreamPartitioner<KO, Void> otherPartitioner,
+ final Function<Optional<Set<Integer>>, Optional<Set<Integer>>>
getPartition
+ ) {
+ this.otherPartitioner = otherPartitioner;
+ this.getPartition = getPartition;
+ }
+
+ @SuppressWarnings("removal")
+ @Override
+ public Optional<Set<Integer>> partitions(final String topic, final KO
key, final SubscriptionWrapper<K> value, final int numPartitions) {
+ return partitions(topic, key, value, new RecordHeaders(),
numPartitions);
+ }
+
+ @Override
+ public Optional<Set<Integer>> partitions(final String topic,
+ final KO key,
+ final SubscriptionWrapper<K>
value,
+ final Headers headers,
+ final int numPartitions) {
+ return getPartition.apply(otherPartitioner.partitions(topic, key,
null, headers, numPartitions));
+ }
+ }
+
+ private static class ForeignResponseSinkPartitioner<K, VO> implements
StreamPartitioner<K, SubscriptionResponseWrapper<VO>> {
+ private final StreamPartitioner<K, Void> partitioner;
+ private final Function<Optional<Set<Integer>>, Optional<Set<Integer>>>
getPartition;
+
+ private ForeignResponseSinkPartitioner(
+ final StreamPartitioner<K, Void> partitioner,
+ final Function<Optional<Set<Integer>>, Optional<Set<Integer>>>
getPartition
+ ) {
+ this.partitioner = partitioner;
+ this.getPartition = getPartition;
+ }
+
+ @SuppressWarnings("removal")
+ @Override
+ public Optional<Set<Integer>> partitions(final String topic, final K
key, final SubscriptionResponseWrapper<VO> value, final int numPartitions) {
+ return partitions(topic, key, value, new RecordHeaders(),
numPartitions);
+ }
+
+ @Override
+ public Optional<Set<Integer>> partitions(final String topic,
+ final K key,
+ final
SubscriptionResponseWrapper<VO> value,
+ final Headers headers,
+ final int numPartitions) {
+ return getPartition.apply(partitioner.partitions(topic, key, null,
headers, numPartitions));
+ }
+ }
+
+ private static class DefaultForeignResponseSinkPartitioner<K, VO>
implements StreamPartitioner<K, SubscriptionResponseWrapper<VO>> {
+ @SuppressWarnings("removal")
+ @Override
+ public Optional<Set<Integer>> partitions(final String topic, final K
key, final SubscriptionResponseWrapper<VO> value, final int numPartitions) {
+ return partitions(topic, key, value, new RecordHeaders(),
numPartitions);
Review Comment:
same
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStreamPartitioner.java:
##########
@@ -32,9 +34,15 @@ public DefaultStreamPartitioner(final Serializer<K>
keySerializer) {
this.keySerializer = keySerializer;
}
+ @SuppressWarnings({"removal"})
@Override
public Optional<Set<Integer>> partitions(final String topic, final K key,
final V value, final int numPartitions) {
- final byte[] keyBytes = keySerializer.serialize(topic, key);
+ return partitions(topic, key, value, new RecordHeaders(),
numPartitions);
+ }
+
+ @Override
+ public Optional<Set<Integer>> partitions(final String topic, final K key,
final V value, final Headers headers, final int numPartitions) {
+ final byte[] keyBytes = keySerializer.serialize(topic, headers, key);
Review Comment:
The issue mentioned above by @aliehsaeedii also applies here right? It's not
limited to `WindowedStreamsPartitioner` as called out on the KIP, but also
non-windowed case if effected.
##########
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java:
##########
@@ -1780,38 +1782,70 @@ 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>
+ * 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.
+ * <p>
+ * See {@link #queryMetadataForKey(String, Object, Headers, Serializer)}
for more details.
Review Comment:
We can remove this line, when we link inline above
##########
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:
This is a little bit of a pickle... An it's not really limited
`WindowedSerializer` (which is an internal interface anyway, while the public
implementations (eg `TimeWindowedSerializer`) are not `final` classes), but
also the actually key-type serializers a `WindowedSerializer` wraps...
##########
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);
Review Comment:
Same question -- this seems to be internal, and thus we should know that
this method is never called?
##########
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java:
##########
@@ -1780,38 +1782,70 @@ 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>
+ * 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.
+ * <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);
+ }
+
/**
* Finds the metadata containing the active hosts and standby hosts where
the key being queried would reside.
*
* @param storeName the {@code storeName} to find metadata for
* @param key the key to find metadata for
+ * @param headers the record headers
* @param keySerializer serializer for the key
* @param <K> key type
* Returns {@link KeyQueryMetadata} containing all metadata about hosting
the given key for the given store,
* or {@code null} if no matching metadata could be found.
*/
public <K> KeyQueryMetadata queryMetadataForKey(final String storeName,
final K key,
+ final Headers headers,
final Serializer<K>
keySerializer) {
validateIsRunningOrRebalancing();
- return streamsMetadataState.keyQueryMetadataForKey(storeName, key,
keySerializer);
+ return streamsMetadataState.keyQueryMetadataForKey(storeName, key,
headers, keySerializer);
+ }
+
+ /**
+ * Finds the metadata containing the active hosts and standby hosts where
the key being queried would reside,
+ * using the supplied partitioner and without requiring record headers to
be provided.
+ * <p>
+ * 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.
+ * <p>
+ * See {@link #queryMetadataForKey(String, Object, Headers,
StreamPartitioner)} for more details.
Review Comment:
As above
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStreamPartitioner.java:
##########
@@ -32,9 +34,15 @@ public DefaultStreamPartitioner(final Serializer<K>
keySerializer) {
this.keySerializer = keySerializer;
}
+ @SuppressWarnings({"removal"})
@Override
public Optional<Set<Integer>> partitions(final String topic, final K key,
final V value, final int numPartitions) {
- final byte[] keyBytes = keySerializer.serialize(topic, key);
+ return partitions(topic, key, value, new RecordHeaders(),
numPartitions);
Review Comment:
Similar. Should we throw?
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java:
##########
@@ -1438,4 +1436,78 @@ private static void maybeSetOutputVersioned(final
GraphNode tableNode,
tableNode.setOutputVersioned(materializedInternal.storeSupplier()
instanceof VersionedBytesStoreSupplier);
}
}
+
+ private static class SubscriptionSinkPartitioner<K, KO> implements
StreamPartitioner<KO, SubscriptionWrapper<K>> {
+ private final StreamPartitioner<KO, Void> otherPartitioner;
+ private final Function<Optional<Set<Integer>>, Optional<Set<Integer>>>
getPartition;
+
+ private SubscriptionSinkPartitioner(
+ final StreamPartitioner<KO, Void> otherPartitioner,
+ final Function<Optional<Set<Integer>>, Optional<Set<Integer>>>
getPartition
+ ) {
+ this.otherPartitioner = otherPartitioner;
+ this.getPartition = getPartition;
+ }
+
+ @SuppressWarnings("removal")
+ @Override
+ public Optional<Set<Integer>> partitions(final String topic, final KO
key, final SubscriptionWrapper<K> value, final int numPartitions) {
+ return partitions(topic, key, value, new RecordHeaders(),
numPartitions);
+ }
+
+ @Override
+ public Optional<Set<Integer>> partitions(final String topic,
+ final KO key,
+ final SubscriptionWrapper<K>
value,
+ final Headers headers,
+ final int numPartitions) {
+ return getPartition.apply(otherPartitioner.partitions(topic, key,
null, headers, numPartitions));
+ }
+ }
+
+ private static class ForeignResponseSinkPartitioner<K, VO> implements
StreamPartitioner<K, SubscriptionResponseWrapper<VO>> {
+ private final StreamPartitioner<K, Void> partitioner;
+ private final Function<Optional<Set<Integer>>, Optional<Set<Integer>>>
getPartition;
+
+ private ForeignResponseSinkPartitioner(
+ final StreamPartitioner<K, Void> partitioner,
+ final Function<Optional<Set<Integer>>, Optional<Set<Integer>>>
getPartition
+ ) {
+ this.partitioner = partitioner;
+ this.getPartition = getPartition;
+ }
+
+ @SuppressWarnings("removal")
+ @Override
+ public Optional<Set<Integer>> partitions(final String topic, final K
key, final SubscriptionResponseWrapper<VO> value, final int numPartitions) {
+ return partitions(topic, key, value, new RecordHeaders(),
numPartitions);
Review Comment:
as above
--
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]
