KAFKA-16362: Fix type-unsafety in KStreamKStreamJoin caused by isLeftSide

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

@@ -172,8 +172,7 @@ public <K, V1, V2, VOut> KStream<K, VOut> join(final KStream<K, V1> lhs,
         final TimeTrackerSupplier sharedTimeTrackerSupplier = new TimeTrackerSupplier();
 
         final JoinWindowsInternal internalWindows = new JoinWindowsInternal(windows);
-        final KStreamKStreamJoin<K, V1, V2, VOut> joinThis = new KStreamKStreamJoin<>(
-            true,
+        final KStreamKStreamLeftJoin<K, V1, V2, VOut> joinThis = new KStreamKStreamLeftJoin<>(
             otherWindowStore.name(),
             internalWindows,
             joiner,
@@ -182,8 +181,7 @@ public <K, V1, V2, VOut> KStream<K, VOut> join(final KStream<K, V1> lhs,
             sharedTimeTrackerSupplier
         );
 
-        final KStreamKStreamJoin<K, V2, V1, VOut> joinOther = new KStreamKStreamJoin<>(
-            false,
+        final KStreamKStreamRightJoin<K, V1, V2, VOut> joinOther = new KStreamKStreamRightJoin<>(
             thisWindowStore.name(),
             internalWindows,
             AbstractStream.reverseJoinerWithKey(joiner),

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

@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.kstream.internals;
+
+import org.apache.kafka.streams.kstream.ValueJoinerWithKey;
+import org.apache.kafka.streams.kstream.internals.KStreamImplJoin.TimeTrackerSupplier;
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.state.internals.LeftOrRightValue;
+import org.apache.kafka.streams.state.internals.TimestampedKeyAndJoinSide;
+
+import java.util.Optional;
+
+class KStreamKStreamLeftJoin<K, VL, VR, VOut> extends KStreamKStreamJoin<K, VL, VR, VOut, VL, VR> {
+
+    KStreamKStreamLeftJoin(final String otherWindowName,
+                           final JoinWindowsInternal windows,
+                           final ValueJoinerWithKey<? super K, ? super VL, ? super VR, ? extends VOut> joiner,
+                           final boolean outer,
+                           final Optional<String> outerJoinWindowName,
+                           final TimeTrackerSupplier sharedTimeTrackerSupplier) {
+        super(otherWindowName, windows, joiner, outer, outerJoinWindowName, windows.beforeMs, windows.afterMs,
+                sharedTimeTrackerSupplier);
+    }
+
+    @Override
+    public Processor<K, VL, K, VOut> get() {
+        return new KStreamKStreamJoinLeftProcessor();
+    }
+
+    private class KStreamKStreamJoinLeftProcessor extends KStreamKStreamJoinProcessor {
+
+
+        @Override
+        public TimestampedKeyAndJoinSide<K> makeThisKey(final K key, final long timestamp) {
+            return TimestampedKeyAndJoinSide.makeLeft(key, timestamp);
+        }
+
+        @Override
+        public LeftOrRightValue<VL, VR> makeThisValue(final VL thisValue) {
+            return LeftOrRightValue.makeLeftValue(thisValue);
+        }
+
+        @Override
+        public TimestampedKeyAndJoinSide<K> makeOtherKey(final K key, final long timestamp) {
+            return TimestampedKeyAndJoinSide.makeRight(key, timestamp);
+        }
+
+        @Override
+        public VL getThisValue(final LeftOrRightValue<? extends VL, ? extends VR> leftOrRightValue) {
+            return leftOrRightValue.getLeftValue();
+        }
+
+        @Override
+        public VR getOtherValue(final LeftOrRightValue<? extends VL, ? extends VR> leftOrRightValue) {
+            return leftOrRightValue.getRightValue();
+        }
+    }
+}

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

@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.kstream.internals;
+
+import org.apache.kafka.streams.kstream.ValueJoinerWithKey;
+import org.apache.kafka.streams.kstream.internals.KStreamImplJoin.TimeTrackerSupplier;
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.state.internals.LeftOrRightValue;
+import org.apache.kafka.streams.state.internals.TimestampedKeyAndJoinSide;
+
+import java.util.Optional;
+
+class KStreamKStreamRightJoin<K, VL, VR, VOut> extends KStreamKStreamJoin<K, VL, VR, VOut, VR, VL> {
+
+    KStreamKStreamRightJoin(final String otherWindowName,
+                            final JoinWindowsInternal windows,
+                            final ValueJoinerWithKey<? super K, ? super VR, ? super VL, ? extends VOut> joiner,
+                            final boolean outer,
+                            final Optional<String> outerJoinWindowName,
+                            final TimeTrackerSupplier sharedTimeTrackerSupplier) {
+        super(otherWindowName, windows, joiner, outer, outerJoinWindowName, windows.afterMs, windows.beforeMs,
+                sharedTimeTrackerSupplier);
+    }
+
+    @Override
+    public Processor<K, VR, K, VOut> get() {
+        return new KStreamKStreamRightJoinProcessor();
+    }
+
+    private class KStreamKStreamRightJoinProcessor extends KStreamKStreamJoinProcessor {
+        @Override
+        public TimestampedKeyAndJoinSide<K> makeThisKey(final K key, final long timestamp) {
+            return TimestampedKeyAndJoinSide.makeRight(key, timestamp);
+        }
+
+        @Override
+        public LeftOrRightValue<VL, VR> makeThisValue(final VR thisValue) {
+            return LeftOrRightValue.makeRightValue(thisValue);
+        }
+
+        @Override
+        public TimestampedKeyAndJoinSide<K> makeOtherKey(final K key, final long timestamp) {
+            return TimestampedKeyAndJoinSide.makeLeft(key, timestamp);
+        }
+
+        @Override
+        public VR getThisValue(final LeftOrRightValue<? extends VL, ? extends VR> leftOrRightValue) {
+            return leftOrRightValue.getRightValue();
+        }
+
+        @Override
+        public VL getOtherValue(final LeftOrRightValue<? extends VL, ? extends VR> leftOrRightValue) {
+            return leftOrRightValue.getLeftValue();
+        }
+    }
+}

streams/src/main/java/org/apache/kafka/streams/state/internals/LeftOrRightValue.java

@@ -63,21 +63,6 @@ public static <V1, V2> LeftOrRightValue<V1, V2> makeRightValue(final V2 rightVal
         return new LeftOrRightValue<>(null, rightValue);
     }
 
-    /**
-     * Create a new {@link LeftOrRightValue} instance with the V value as {@code leftValue} if
-     * {@code isLeftSide} is True; otherwise {@code rightValue} if {@code isLeftSide} is False.
-     *
-     * @param value the V value (either V1 or V2 type)
-     * @param <V>   the type of the value
-     * @return a new {@link LeftOrRightValue} instance
-     */
-    public static <V> LeftOrRightValue make(final boolean isLeftSide, final V value) {
-        Objects.requireNonNull(value, "value is null");
-        return isLeftSide
-            ? LeftOrRightValue.makeLeftValue(value)
-            : LeftOrRightValue.makeRightValue(value);
-    }
-
     public V1 getLeftValue() {
         return leftValue;
     }

streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedKeyAndJoinSide.java

@@ -42,17 +42,25 @@ private TimestampedKeyAndJoinSide(final boolean leftSide, final K key, final lon
     }
 
     /**
-     * Create a new {@link TimestampedKeyAndJoinSide} instance if the provide {@code key} is not {@code null}.
+     * Create a new {@link TimestampedKeyAndJoinSide} instance for the left join side if the provide {@code key} is not {@code null}.
      *
-     * @param leftSide True if the key is part of the left join side; False if it is from the right join side
      * @param key      the key
      * @param <K>      the type of the key
-     * @return a new {@link TimestampedKeyAndJoinSide} instance if the provide {@code key} is not {@code null}
+     * @return a new {@link TimestampedKeyAndJoinSide} instance for the left join side if the provide {@code key} is not {@code null}
      */
-    public static <K> TimestampedKeyAndJoinSide<K> make(final boolean leftSide, final K key, final long timestamp) {
-        return new TimestampedKeyAndJoinSide<>(leftSide, key, timestamp);
+    public static <K> TimestampedKeyAndJoinSide<K> makeLeft(final K key, final long timestamp) {
+        return new TimestampedKeyAndJoinSide<>(true, key, timestamp);
+    }
+    /**
+     * Create a new {@link TimestampedKeyAndJoinSide} instance for the right join side if the provide {@code key} is not {@code null}.
+     *
+     * @param key      the key
+     * @param <K>      the type of the key
+     * @return a new {@link TimestampedKeyAndJoinSide} instance for the right join side if the provide {@code key} is not {@code null}
+     */
+    public static <K> TimestampedKeyAndJoinSide<K> makeRight(final K key, final long timestamp) {
+        return new TimestampedKeyAndJoinSide<>(false, key, timestamp);
     }
-
     public boolean isLeftSide() {
         return leftSide;
     }
@@ -89,4 +97,4 @@ public boolean equals(final Object o) {
     public int hashCode() {
         return Objects.hash(leftSide, key, timestamp);
     }
-}
+}

streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedKeyAndJoinSideDeserializer.java

@@ -55,11 +55,12 @@ public void configure(final Map<String, ?> configs, final boolean isKey) {
 
     @Override
     public TimestampedKeyAndJoinSide<K> deserialize(final String topic, final byte[] data) {
-        final boolean bool = data[StateSerdes.TIMESTAMP_SIZE] == 1;
+        final boolean isLeft = data[StateSerdes.TIMESTAMP_SIZE] == 1;
         final K key = keyDeserializer.deserialize(topic, rawKey(data));
         final long timestamp = timestampDeserializer.deserialize(topic, rawTimestamp(data));
 
-        return TimestampedKeyAndJoinSide.make(bool, key, timestamp);
+        return isLeft ? TimestampedKeyAndJoinSide.makeLeft(key, timestamp) :
+                TimestampedKeyAndJoinSide.makeRight(key, timestamp);
     }
 
     private byte[] rawTimestamp(final byte[] data) {

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoinTest.java

@@ -456,8 +456,7 @@ public void shouldThrottleEmitNonJoinedOuterRecordsEvenWhenClockDrift() {
          * This test is testing something internal to [[KStreamKStreamJoin]], so we had to setup low-level api manually.
          */
         final KStreamImplJoin.TimeTrackerSupplier tracker = new KStreamImplJoin.TimeTrackerSupplier();
-        final KStreamKStreamJoin<String, String, String, String> join = new KStreamKStreamJoin<>(
-                false,
+        final KStreamKStreamRightJoin<String, String, String, String> join = new KStreamKStreamRightJoin<>(
                 "other",
                 new JoinWindowsInternal(JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(1000))),
                 (key, v1, v2) -> v1 + v2,

streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedKeyAndJoinSideSerializerTest.java

@@ -34,7 +34,7 @@ public class TimestampedKeyAndJoinSideSerializerTest {
     public void shouldSerializeKeyWithJoinSideAsTrue() {
         final String value = "some-string";
 
-        final TimestampedKeyAndJoinSide<String> timestampedKeyAndJoinSide = TimestampedKeyAndJoinSide.make(true, value, 10);
+        final TimestampedKeyAndJoinSide<String> timestampedKeyAndJoinSide = TimestampedKeyAndJoinSide.makeLeft(value, 10);
 
         final byte[] serialized =
             STRING_SERDE.serializer().serialize(TOPIC, timestampedKeyAndJoinSide);
@@ -51,7 +51,7 @@ public void shouldSerializeKeyWithJoinSideAsTrue() {
     public void shouldSerializeKeyWithJoinSideAsFalse() {
         final String value = "some-string";
 
-        final TimestampedKeyAndJoinSide<String> timestampedKeyAndJoinSide = TimestampedKeyAndJoinSide.make(false, value, 20);
+        final TimestampedKeyAndJoinSide<String> timestampedKeyAndJoinSide = TimestampedKeyAndJoinSide.makeRight(value, 20);
 
         final byte[] serialized =
             STRING_SERDE.serializer().serialize(TOPIC, timestampedKeyAndJoinSide);
@@ -67,6 +67,6 @@ public void shouldSerializeKeyWithJoinSideAsFalse() {
     @Test
     public void shouldThrowIfSerializeNullData() {
         assertThrows(NullPointerException.class,
-            () -> STRING_SERDE.serializer().serialize(TOPIC, TimestampedKeyAndJoinSide.make(true, null, 0)));
+            () -> STRING_SERDE.serializer().serialize(TOPIC, TimestampedKeyAndJoinSide.makeLeft(null, 0)));
     }
 }