You are viewing an old version of this page. View the current version.

Compare with Current View Page History

« Previous Version 8 Next »

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 methodsCurrent argument typeNew argument type
(KGroupedStream|KGroupedTable).aggregateAggregator<K, V, T>Aggregator<? super K, ? super V, T>
(KTable|KStream).filter*, KStream.branchPredicate<K, V> Predicate<? super K, ? super V>
(KStream|KTable).groupBy, KStream.(selectKey|map|flatMap), KTable.toStreamKeyValueMapper<K, V, X>KeyValueMapper<? super K, ? super V, X>
(KStream|KTable).mapValues, KStream.flatMapValuesValueMapper<V, X>ValueMapper<? super V, ? extends X>
KStream.transformTransformerSupplier<K, V, X>TransformerSupplier<? super K, ? super V, ? extends X>

KStream.transformValues

ValueTransformerSupplier<V, X>ValueTransformerSupplier<? super V, ? extends X>
(KStream|Ktable).foreachForeachAction<K, V>ForeachAction<? super K, ? super V>

KStream.process

ProcessorSupplier<K, V>ProcessorSupplier<? super K, ? super V>
(KStream|KTable).*joinValueJoiner<K, V, R>ValueJoiner<? super K, ? super V, ? extends R>

(KStream|KTable).(to|through)

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

  • No labels