Status
Current state: Under Discussion
Discussion thread: here
JIRA: KAFKA-4481
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Motivation
Several Kafka Streams methods currently take arguments that are functions parameterized in the key and value types to apply various transformations to KStreams and KTables. Those functions are currently invariant in their input in output types, when they should probably be contravariant in their key in value input types, and covariant in their result type.
For instance, KStream<K, V>.filter(Predicate<K, V> predicate)
should be KStream.filter(Predicate<? super K, ? super V> predicate)
to accept predicates that can act on any supertype of K
, or V
. More concretely, if Cat
extends Animal
, and I have Predicate<Animal, Object> animalPredicate
, then I should be able to call KStream<Cat, Picture>.filter(animalPredicate)
Conversely for result types, KStream<K, V>.map(ValueMapper<V, R> mapper)
should be KStream<K, V>.map(ValueMapper<? super V, ? extends R> mapper)
. For example I can apply ValueTransformer<Object, String> toStringTransformer
to KStream<K, Serializable>.map(toStringTransformer)
and the result can safely be used as either KStream<K, String>
or as KStream<K, Serializable>
without relying on unchecked casts.
This change will make it easier to write reusable code for transformations, without requiring additional wrappers around existing code, or the unnecessary use of unchecked casts.
The same reasoning applies to the key, value and result types defined in methods that take Aggregator
, StreamPartitioner
, KeyValueMapper
, ValueMapper
, ProcessorSupplier
, TransformerSupplier
, ValueTransformerSupplier
, ForeachAction
, StreamPartitioner
, and ValueJoiner
.
Public Interfaces
Affected methods | Current argument type | New argument type |
---|---|---|
(KGroupedStream|KGroupedTable).aggregate | Aggregator<K, V, T> | Aggregator<? super K, ? super V, T> |
(KTable|KStream).filter*, KStream.branch | Predicate<K, V> | Predicate<? super K, ? super V> |
(KStream|KTable).groupBy, KStream.(selectKey|map|flatMap), KTable.toStream | KeyValueMapper<K, V, X> | KeyValueMapper<? super K, ? super V, X> |
(KStream|KTable).mapValues, KStream.flatMapValues | ValueMapper<V, X> | ValueMapper<? super V, ? extends X> |
KStream.transform | TransformerSupplier<K, V, X> | TransformerSupplier<? super K, ? super V, ? extends X> |
| ValueTransformerSupplier<V, X> | ValueTransformerSupplier<? super V, ? extends X> |
(KStream|Ktable).foreach | ForeachAction<K, V> | ForeachAction<? super K, ? super V> |
| ProcessorSupplier<K, V> | ProcessorSupplier<? super K, ? super V> |
(KStream|KTable).*join | ValueJoiner<K, V, R> | ValueJoiner<? super K, ? super V, ? extends R> |
| StreamPartitioner<K, V> | StreamPartitioner<? super K, ? super V> |
KafkaStreams.metadataForKey | StreamPartitioner<K, V> | StreamPartitioner<? super K, ? super V> |
Proposed Changes
This KIP proposes changing the methods on the interfaces listed above to relax function arguments parameterized in key in value types to accept super-types of those key and values.
Compatibility, Deprecation, and Migration Plan
- This change is binary compatible
- This change is source compatible for anyone merely calling the existing APIs
- This change is not source compatible for anyone extending the affected classes / interfaces.
Rejected Alternatives
None