KAFKA-18644: improve generic type names for KStreamImpl and KTableImpl #18722

mjsax · 2025-01-28T03:28:13Z

No description provided.

mjsax · 2025-01-28T03:29:01Z

streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java

@@ -3062,7 +3062,7 @@ <GK, GV, RV> KStream<K, RV> leftJoin(final GlobalKTable<GK, GV> globalTable,
     * @see #map(KeyValueMapper)
     */
    <KOut, VOut> KStream<KOut, VOut> process(
-        final ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier,
+        final ProcessorSupplier<? super K, ? super V, ? extends KOut, ? extends VOut> processorSupplier,


While this is technically public API, to me it's a bug-fix, so would not want to do a KIP for it... Similar to KTable interface changes. -- On the other hand, it more than a one-liner... Overall not sure...

We did a KIP back in the days: https://cwiki.apache.org/confluence/display/KAFKA/KIP-100+-+Relax+Type+constraints+in+Kafka+Streams+API

Note, that some interfaces actually do use ? super/extend X even if the original KIP adding the methods does not say so... So we did diverge from it effectively already.

Thoughts?

I think given that we have current usage of ? super/extend X in the code base, we don't need a KIP.

Yeah, this should be covered by KIP-100 in spirit, I don't think you need a KIP.

bbejeck

Thanks for the PR @mjsax , overall looks good to me - just a couple of questions

bbejeck · 2025-01-28T23:58:13Z

streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java

@@ -3062,7 +3062,7 @@ <GK, GV, RV> KStream<K, RV> leftJoin(final GlobalKTable<GK, GV> globalTable,
     * @see #map(KeyValueMapper)
     */
    <KOut, VOut> KStream<KOut, VOut> process(
-        final ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier,
+        final ProcessorSupplier<? super K, ? super V, ? extends KOut, ? extends VOut> processorSupplier,


I think given that we have current usage of ? super/extend X in the code base, we don't need a KIP.

bbejeck · 2025-01-29T00:12:16Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/AbstractStream.java

        Objects.requireNonNull(valueMapper, "valueMapper can't be null");
        return (readOnlyKey, value) -> valueMapper.apply(value);
    }

-    static <K, V1, V2, VR> ValueJoinerWithKey<K, V1, V2, VR> toValueJoinerWithKey(final ValueJoiner<V1, V2, VR> valueJoiner) {
+    static <K, VLeft, VRight, VOut> ValueJoinerWithKey<K, VLeft, VRight, VOut> toValueJoinerWithKey(final ValueJoiner<VLeft, VRight, VOut> valueJoiner) {


All the other methods have generics of ? super or ? extends why not here - asking more more my own education.

We don't need it from my understanding (please correct me if I am wrong).

Assume, we have KStreams<K, A> and KStream<K, B> that we join, to result KStream<K, C>. (All K,A,B,C are concrete types)

We use ValueJoiner<KFoo, V, W, X> (which are concrete types) with K extend KFoo, A extends V, B extends W and (reverser for output), X extends C, we just want to translate the ValueJoiner to <KFoo, W, V, X> which are still concrete types. And yes, these concreate types must match ? extends... and ? super for the KStreams we process, but they still do.

Thus, we don't need this helper to be able to translate a child/super type of ValueJoiner<KFoo, V, W, X> -- it's sufficient to just translate the concrete types <KFoo, V, W, X>.

Or do I get this wrong?

Nope, that sounds good to me, thanks for the clarification.

bbejeck · 2025-01-29T00:52:40Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java

@@ -153,9 +153,9 @@ public KStream<K, V> filter(final Predicate<? super K, ? super V> predicate,
        Objects.requireNonNull(named, "named can't be null");

        final String name = new NamedInternal(named).orElseGenerateWithPrefix(builder, FILTER_NAME);
-        final ProcessorParameters<? super K, ? super V, ?, ?> processorParameters =
+        final ProcessorParameters<K, V, K, V> processorParameters =


why the changes to the generic types? For example, from ? super K to K

We need ? super K only for the user-implemented interfaces / UDFs. This allows to process a KStream of type <K,V>, with a UDF which accepts a supertype of K and or V (or returns child type, eg, ? extend VOut)

However, it does not change the type of the input KStream itself -- and also not the type of the result KStream... The actual/concrete type is still <K,V>.

Assume we filter a KStream<Void,Cat> and apply an Predicate<Void,Animal> which gives us animals with black fur. The result KStream is still <Void,Cat>...

If we would map KStream<Void,Something> using ValueMapper<Cats>, we can still treat the result as KStream<Void,Animal>.

Thus, for the internal ProcessParameter we only need to track the actual/concrete types, which are independent of potential super/child type for UDF generics.

The code compiles both ways for ProcessorParameter, but it's just "silly" to use anything except the concrete types, as the compiler cannot check anything else anyway....

bbejeck · 2025-01-29T00:59:25Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java

@@ -248,9 +248,9 @@ public <KR, VR> KStream<KR, VR> map(final KeyValueMapper<? super K, ? super V, ?
        Objects.requireNonNull(named, "named can't be null");

        final String name = new NamedInternal(named).orElseGenerateWithPrefix(builder, MAP_NAME);
-        final ProcessorParameters<? super K, ? super V, ?, ?> processorParameters =
+        final ProcessorParameters<K, V, KR, VR> processorParameters =


And here from ? to KR

For map() we know that the output types are KR and VR -- so why should we leave them unchecked as ? -- we know the types and the compiler can verify them for us.

I believe, we did use ? to work with the old PAPI which did not have output types on Processor interface, but know we can improve type safety by enforcing the check now, as we have the new api.Processor interface in place.

bbejeck · 2025-01-29T01:10:21Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java

-                                                                                             final BaseRepartitionNodeBuilder<K1, V1, RN> baseRepartitionNodeBuilder) {
-
+    static <KStream, VStream, RepartitionNode extends BaseRepartitionNode<KStream, VStream>> String createRepartitionedSource(
+        final InternalStreamsBuilder builder,


why KStream and VStream? Are there two streams involved here? It's been a while since I've looked at this part of the code.

Oh. I see the confusing. My bad. Not a good name... I just meant Key type of a Stream... K1 is just terrible...

Re-using <K,V> even if possible, would imply shadowing the KStream<K,V> generics... (what might be ok, as we don't need access to them, but it's not recommended in general, but we should use a different name.

Any ideas for a good name?

Updating to Key and Value for now. Happy to pick something else if you have a good idea.

sounds good to me, thanks for clearing that up - and yeah naming is hard.

Yes. Cf #18721 (comment) -- Same question... Any ideas?

We cannot just use Key, and Value there... 🤔

bbejeck · 2025-01-29T01:12:19Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java

-                                                        final ValueJoinerWithKey<? super K, ? super V, ? super VG, ? extends VR> joiner,
-                                                        final boolean leftJoin,
-                                                        final Named named) {
+    private <KGlobalTable, VGlobalTable, VOut> KStream<K, VOut> globalTableJoin(


nit would GlobalTableK be more descriptive than KGlobalTable ? same for VGlobalTable

Guess it's the same question as above -- we might want to use some different name to begin with. Any ideas?

Updating to GlobalKey and GlobalValue for now. Happy to pick something else if you have a good idea.

bbejeck

Thanks for the comment responses. LGTM pending tests passing

mjsax · 2025-01-30T17:45:21Z

Java 23 passed.

Java 17:

Found 1 test failures:
FAILED ❌ KafkaAdminClientTest > testAdminClientApisAuthenticationFailure()
Found 5 flaky test failures:
FLAKY ⚠️  KafkaShareConsumerTest > testVerifyFetchAndCloseImplicit()
FLAKY ⚠️  AbstractCoordinatorTest > testWakeupAfterSyncGroupReceivedExternalCompletion()
FLAKY ⚠️  AbstractCoordinatorTest > testWakeupAfterSyncGroupReceived()
FLAKY ⚠️  StickyAssignorTest > testLargeAssignmentAndGroupWithUniformSubscription(boolean).hasConsumerRack = false
FLAKY ⚠️  PlaintextAdminIntegrationTest > testConsumerGroups(String, String).quorum=kraft.groupProtocol=consumer

mjsax · 2025-01-30T17:46:06Z

Rebased to resolve conflicts. Will merge after CI finished.

mjsax added the streams label Jan 28, 2025

mjsax commented Jan 28, 2025

View reviewed changes

mjsax force-pushed the kafka-18644-dls-type-safely-kstream-ktable branch from 7abb52a to 153e0b4 Compare January 28, 2025 03:35

bbejeck reviewed Jan 29, 2025

View reviewed changes

bbejeck approved these changes Jan 29, 2025

View reviewed changes

mjsax added 4 commits January 30, 2025 09:44

KAFKA-18644: improve generic type names for KStreamImpl and KTableImpl

21c8e5b

more fixed

0298f80

fixes for KTableImpl

ac862e7

github comments

f6364b6

mjsax force-pushed the kafka-18644-dls-type-safely-kstream-ktable branch from e83ac66 to f6364b6 Compare January 30, 2025 17:46

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-18644: improve generic type names for KStreamImpl and KTableImpl #18722

KAFKA-18644: improve generic type names for KStreamImpl and KTableImpl #18722

mjsax commented Jan 28, 2025

mjsax Jan 28, 2025 •

edited

Loading

bbejeck Jan 28, 2025

lucasbru Jan 30, 2025

bbejeck left a comment

bbejeck Jan 28, 2025

bbejeck Jan 29, 2025

mjsax Jan 29, 2025

bbejeck Jan 29, 2025

bbejeck Jan 29, 2025

mjsax Jan 29, 2025

bbejeck Jan 29, 2025

mjsax Jan 29, 2025 •

edited

Loading

bbejeck Jan 29, 2025

mjsax Jan 29, 2025

mjsax Jan 29, 2025

bbejeck Jan 29, 2025

mjsax Jan 29, 2025

bbejeck Jan 29, 2025

mjsax Jan 29, 2025

mjsax Jan 29, 2025

bbejeck left a comment

mjsax commented Jan 30, 2025

mjsax commented Jan 30, 2025

KAFKA-18644: improve generic type names for KStreamImpl and KTableImpl #18722

Are you sure you want to change the base?

KAFKA-18644: improve generic type names for KStreamImpl and KTableImpl #18722

Conversation

mjsax commented Jan 28, 2025

mjsax Jan 28, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax Jan 29, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck left a comment

Choose a reason for hiding this comment

mjsax commented Jan 30, 2025

mjsax commented Jan 30, 2025

mjsax Jan 28, 2025 •

edited

Loading

mjsax Jan 29, 2025 •

edited

Loading