diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKey.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKey.java
index 016f24cb4a055..dda47a2181bbd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKey.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKey.java
@@ -14,26 +14,25 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.apache.kafka.streams.kstream.internals.foreignkeyjoin;
 
 import java.util.Objects;
 
-public class CombinedKey<KF, KP> {
-    private final KF foreignKey;
-    private final KP primaryKey;
+public class CombinedKey<KRight, KLeft> {
+    private final KRight foreignKey;
+    private final KLeft primaryKey;
 
-    CombinedKey(final KF foreignKey, final KP primaryKey) {
+    CombinedKey(final KRight foreignKey, final KLeft primaryKey) {
         Objects.requireNonNull(primaryKey, "primaryKey can't be null");
         this.foreignKey = foreignKey;
         this.primaryKey = primaryKey;
     }
 
-    public KF foreignKey() {
+    public KRight foreignKey() {
         return foreignKey;
     }
 
-    public KP primaryKey() {
+    public KLeft primaryKey() {
         return primaryKey;
     }
 
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKeySchema.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKeySchema.java
index 6b735f5b56a01..176205c724870 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKeySchema.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKeySchema.java
@@ -28,20 +28,20 @@
 /**
  * Factory for creating CombinedKey serializers / deserializers.
  */
-public class CombinedKeySchema<KO, K> {
+public class CombinedKeySchema<KRight, KLeft> {
     private final Supplier<String> undecoratedPrimaryKeySerdeTopicSupplier;
     private final Supplier<String> undecoratedForeignKeySerdeTopicSupplier;
     private String primaryKeySerdeTopic;
     private String foreignKeySerdeTopic;
-    private Serializer<K> primaryKeySerializer;
-    private Deserializer<K> primaryKeyDeserializer;
-    private Serializer<KO> foreignKeySerializer;
-    private Deserializer<KO> foreignKeyDeserializer;
+    private Serializer<KLeft> primaryKeySerializer;
+    private Deserializer<KLeft> primaryKeyDeserializer;
+    private Serializer<KRight> foreignKeySerializer;
+    private Deserializer<KRight> foreignKeyDeserializer;
 
     public CombinedKeySchema(final Supplier<String> foreignKeySerdeTopicSupplier,
-                             final Serde<KO> foreignKeySerde,
+                             final Serde<KRight> foreignKeySerde,
                              final Supplier<String> primaryKeySerdeTopicSupplier,
-                             final Serde<K> primaryKeySerde) {
+                             final Serde<KLeft> primaryKeySerde) {
         undecoratedPrimaryKeySerdeTopicSupplier = primaryKeySerdeTopicSupplier;
         undecoratedForeignKeySerdeTopicSupplier = foreignKeySerdeTopicSupplier;
         primaryKeySerializer = primaryKeySerde == null ? null : primaryKeySerde.serializer();
@@ -50,17 +50,17 @@ public CombinedKeySchema(final Supplier<String> foreignKeySerdeTopicSupplier,
         foreignKeySerializer = foreignKeySerde == null ? null : foreignKeySerde.serializer();
     }
 
-    @SuppressWarnings("unchecked")
+    @SuppressWarnings({"unchecked", "resource"})
     public void init(final ProcessorContext<?, ?> context) {
         primaryKeySerdeTopic = undecoratedPrimaryKeySerdeTopicSupplier.get();
         foreignKeySerdeTopic = undecoratedForeignKeySerdeTopicSupplier.get();
-        primaryKeySerializer = primaryKeySerializer == null ? (Serializer<K>) context.keySerde().serializer() : primaryKeySerializer;
-        primaryKeyDeserializer = primaryKeyDeserializer == null ? (Deserializer<K>) context.keySerde().deserializer() : primaryKeyDeserializer;
-        foreignKeySerializer = foreignKeySerializer == null ? (Serializer<KO>) context.keySerde().serializer() : foreignKeySerializer;
-        foreignKeyDeserializer = foreignKeyDeserializer == null ? (Deserializer<KO>) context.keySerde().deserializer() : foreignKeyDeserializer;
+        primaryKeySerializer = primaryKeySerializer == null ? (Serializer<KLeft>) context.keySerde().serializer() : primaryKeySerializer;
+        primaryKeyDeserializer = primaryKeyDeserializer == null ? (Deserializer<KLeft>) context.keySerde().deserializer() : primaryKeyDeserializer;
+        foreignKeySerializer = foreignKeySerializer == null ? (Serializer<KRight>) context.keySerde().serializer() : foreignKeySerializer;
+        foreignKeyDeserializer = foreignKeyDeserializer == null ? (Deserializer<KRight>) context.keySerde().deserializer() : foreignKeyDeserializer;
     }
 
-    Bytes toBytes(final KO foreignKey, final K primaryKey) {
+    Bytes toBytes(final KRight foreignKey, final KLeft primaryKey) {
         //The serialization format - note that primaryKeySerialized may be null, such as when a prefixScan
         //key is being created.
         //{Integer.BYTES foreignKeyLength}{foreignKeySerialized}{Optional-primaryKeySerialized}
@@ -79,22 +79,22 @@ Bytes toBytes(final KO foreignKey, final K primaryKey) {
     }
 
 
-    public CombinedKey<KO, K> fromBytes(final Bytes data) {
+    public CombinedKey<KRight, KLeft> fromBytes(final Bytes data) {
         //{Integer.BYTES foreignKeyLength}{foreignKeySerialized}{Optional-primaryKeySerialized}
         final byte[] dataArray = data.get();
         final ByteBuffer dataBuffer = ByteBuffer.wrap(dataArray);
         final int foreignKeyLength = dataBuffer.getInt();
         final byte[] foreignKeyRaw = new byte[foreignKeyLength];
         dataBuffer.get(foreignKeyRaw, 0, foreignKeyLength);
-        final KO foreignKey = foreignKeyDeserializer.deserialize(foreignKeySerdeTopic, foreignKeyRaw);
+        final KRight foreignKey = foreignKeyDeserializer.deserialize(foreignKeySerdeTopic, foreignKeyRaw);
 
         final byte[] primaryKeyRaw = new byte[dataArray.length - foreignKeyLength - Integer.BYTES];
         dataBuffer.get(primaryKeyRaw, 0, primaryKeyRaw.length);
-        final K primaryKey = primaryKeyDeserializer.deserialize(primaryKeySerdeTopic, primaryKeyRaw);
+        final KLeft primaryKey = primaryKeyDeserializer.deserialize(primaryKeySerdeTopic, primaryKeyRaw);
         return new CombinedKey<>(foreignKey, primaryKey);
     }
 
-    Bytes prefixBytes(final KO key) {
+    Bytes prefixBytes(final KRight key) {
         //The serialization format. Note that primaryKeySerialized is not required/used in this function.
         //{Integer.BYTES foreignKeyLength}{foreignKeySerialized}{Optional-primaryKeySerialized}
 
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignKeyExtractor.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignKeyExtractor.java
index 1d422d26fd6b5..9f597bad4b252 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignKeyExtractor.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignKeyExtractor.java
@@ -30,19 +30,19 @@
  *   <li>{@link #fromBiFunction(BiFunction)} - when the foreign key depends on both key and value</li>
  * </ul>
  *
- * @param <K>  Type of primary table's key
- * @param <V>  Type of primary table's value
- * @param <KO> Type of the foreign key to extract
+ * @param <KLeft>  Type of primary table's key
+ * @param <VLeft>  Type of primary table's value
+ * @param <KRight> Type of the foreign key to extract
  */
 @FunctionalInterface
-public interface ForeignKeyExtractor<K, V, KO> {
-    KO extract(K key, V value);
+public interface ForeignKeyExtractor<KLeft, VLeft, KRight> {
+    KRight extract(KLeft key, VLeft value);
 
-    static <K, V, KO> ForeignKeyExtractor<? super K, ? super V, ? extends KO> fromFunction(Function<? super V, ? extends KO> function) {
+    static <KLeft, VLeft, KRight> ForeignKeyExtractor<? super KLeft, ? super VLeft, ? extends KRight> fromFunction(Function<? super VLeft, ? extends KRight> function) {
         return (key, value) -> function.apply(value);
     }
 
-    static <K, V, KO> ForeignKeyExtractor<? super K, ? super V, ? extends KO> fromBiFunction(BiFunction<? super K, ? super V, ? extends KO> biFunction) {
+    static <KLeft, VLeft, KRight> ForeignKeyExtractor<? super KLeft, ? super VLeft, ? extends KRight> fromBiFunction(BiFunction<? super KLeft, ? super VLeft, ? extends KRight> biFunction) {
         return biFunction::apply;
     }
 }
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignTableJoinProcessorSupplier.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignTableJoinProcessorSupplier.java
index a8b8228ed552f..99d7b22dedc80 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignTableJoinProcessorSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignTableJoinProcessorSupplier.java
@@ -14,7 +14,6 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.apache.kafka.streams.kstream.internals.foreignkeyjoin;
 
 import org.apache.kafka.common.metrics.Sensor;
@@ -43,15 +42,16 @@
 import java.util.Collections;
 import java.util.Set;
 
-public class ForeignTableJoinProcessorSupplier<K, KO, VO> implements
-    ProcessorSupplier<KO, Change<VO>, K, SubscriptionResponseWrapper<VO>> {
+public class ForeignTableJoinProcessorSupplier<KLeft, KRight, VRight>
+    implements ProcessorSupplier<KRight, Change<VRight>, KLeft, SubscriptionResponseWrapper<VRight>> {
+
     private static final Logger LOG = LoggerFactory.getLogger(ForeignTableJoinProcessorSupplier.class);
     private final StoreFactory subscriptionStoreFactory;
-    private final CombinedKeySchema<KO, K> keySchema;
+    private final CombinedKeySchema<KRight, KLeft> keySchema;
     private boolean useVersionedSemantics = false;
 
     public ForeignTableJoinProcessorSupplier(final StoreFactory subscriptionStoreFactory,
-                                             final CombinedKeySchema<KO, K> keySchema) {
+                                             final CombinedKeySchema<KRight, KLeft> keySchema) {
         this.subscriptionStoreFactory = subscriptionStoreFactory;
         this.keySchema = keySchema;
     }
@@ -62,7 +62,7 @@ public Set<StoreBuilder<?>> stores() {
     }
 
     @Override
-    public Processor<KO, Change<VO>, K, SubscriptionResponseWrapper<VO>> get() {
+    public Processor<KRight, Change<VRight>, KLeft, SubscriptionResponseWrapper<VRight>> get() {
         return new KTableKTableJoinProcessor();
     }
 
@@ -75,12 +75,12 @@ public boolean isUseVersionedSemantics() {
         return useVersionedSemantics;
     }
 
-    private final class KTableKTableJoinProcessor extends ContextualProcessor<KO, Change<VO>, K, SubscriptionResponseWrapper<VO>> {
+    private final class KTableKTableJoinProcessor extends ContextualProcessor<KRight, Change<VRight>, KLeft, SubscriptionResponseWrapper<VRight>> {
         private Sensor droppedRecordsSensor;
-        private TimestampedKeyValueStore<Bytes, SubscriptionWrapper<K>> subscriptionStore;
+        private TimestampedKeyValueStore<Bytes, SubscriptionWrapper<KLeft>> subscriptionStore;
 
         @Override
-        public void init(final ProcessorContext<K, SubscriptionResponseWrapper<VO>> context) {
+        public void init(final ProcessorContext<KLeft, SubscriptionResponseWrapper<VRight>> context) {
             super.init(context);
             final InternalProcessorContext<?, ?> internalProcessorContext = (InternalProcessorContext<?, ?>) context;
             droppedRecordsSensor = TaskMetrics.droppedRecordsSensor(
@@ -92,7 +92,7 @@ public void init(final ProcessorContext<K, SubscriptionResponseWrapper<VO>> cont
         }
 
         @Override
-        public void process(final Record<KO, Change<VO>> record) {
+        public void process(final Record<KRight, Change<VRight>> record) {
             // if the key is null, we do not need to proceed aggregating
             // the record with the table
             if (record.key() == null) {
@@ -122,14 +122,14 @@ public void process(final Record<KO, Change<VO>> record) {
             final Bytes prefixBytes = keySchema.prefixBytes(record.key());
 
             //Perform the prefixScan and propagate the results
-            try (final KeyValueIterator<Bytes, ValueAndTimestamp<SubscriptionWrapper<K>>> prefixScanResults =
+            try (final KeyValueIterator<Bytes, ValueAndTimestamp<SubscriptionWrapper<KLeft>>> prefixScanResults =
                      subscriptionStore.range(prefixBytes, Bytes.increment(prefixBytes))) {
 
                 while (prefixScanResults.hasNext()) {
-                    final KeyValue<Bytes, ValueAndTimestamp<SubscriptionWrapper<K>>> next = prefixScanResults.next();
+                    final KeyValue<Bytes, ValueAndTimestamp<SubscriptionWrapper<KLeft>>> next = prefixScanResults.next();
                     // have to check the prefix because the range end is inclusive :(
                     if (prefixEquals(next.key.get(), prefixBytes.get())) {
-                        final CombinedKey<KO, K> combinedKey = keySchema.fromBytes(next.key);
+                        final CombinedKey<KRight, KLeft> combinedKey = keySchema.fromBytes(next.key);
                         context().forward(
                             record.withKey(combinedKey.primaryKey())
                                 .withValue(new SubscriptionResponseWrapper<>(
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ResponseJoinProcessorSupplier.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ResponseJoinProcessorSupplier.java
index 42e9c1cca45ab..a81700b6c0165 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ResponseJoinProcessorSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ResponseJoinProcessorSupplier.java
@@ -14,7 +14,6 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.apache.kafka.streams.kstream.internals.foreignkeyjoin;
 
 import org.apache.kafka.common.errors.UnsupportedVersionException;
@@ -43,23 +42,25 @@
  * of the primary key. This eliminates race-condition results for rapidly-changing foreign-keys for a given primary key.
  * Applies the join and emits nulls according to LEFT/INNER rules.
  *
- * @param <K> Type of primary keys
- * @param <V> Type of primary values
- * @param <VO> Type of foreign values
- * @param <VR> Type of joined result of primary and foreign values
+ * @param <KLeft> Type of primary keys
+ * @param <VLeft> Type of primary values
+ * @param <VRight> Type of foreign values
+ * @param <VOut> Type of joined result of primary and foreign values
  */
-public class ResponseJoinProcessorSupplier<K, V, VO, VR> implements ProcessorSupplier<K, SubscriptionResponseWrapper<VO>, K, VR> {
+public class ResponseJoinProcessorSupplier<KLeft, VLeft, VRight, VOut>
+    implements ProcessorSupplier<KLeft, SubscriptionResponseWrapper<VRight>, KLeft, VOut> {
+
     private static final Logger LOG = LoggerFactory.getLogger(ResponseJoinProcessorSupplier.class);
-    private final KTableValueGetterSupplier<K, V> valueGetterSupplier;
-    private final Serializer<V> constructionTimeValueSerializer;
+    private final KTableValueGetterSupplier<KLeft, VLeft> valueGetterSupplier;
+    private final Serializer<VLeft> constructionTimeValueSerializer;
     private final Supplier<String> valueHashSerdePseudoTopicSupplier;
-    private final ValueJoiner<? super V, ? super VO, ? extends VR> joiner;
+    private final ValueJoiner<? super VLeft, ? super VRight, ? extends VOut> joiner;
     private final boolean leftJoin;
 
-    public ResponseJoinProcessorSupplier(final KTableValueGetterSupplier<K, V> valueGetterSupplier,
-                                         final Serializer<V> valueSerializer,
+    public ResponseJoinProcessorSupplier(final KTableValueGetterSupplier<KLeft, VLeft> valueGetterSupplier,
+                                         final Serializer<VLeft> valueSerializer,
                                          final Supplier<String> valueHashSerdePseudoTopicSupplier,
-                                         final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
+                                         final ValueJoiner<? super VLeft, ? super VRight, ? extends VOut> joiner,
                                          final boolean leftJoin) {
         this.valueGetterSupplier = valueGetterSupplier;
         constructionTimeValueSerializer = valueSerializer;
@@ -69,24 +70,24 @@ public ResponseJoinProcessorSupplier(final KTableValueGetterSupplier<K, V> value
     }
 
     @Override
-    public Processor<K, SubscriptionResponseWrapper<VO>, K, VR> get() {
+    public Processor<KLeft, SubscriptionResponseWrapper<VRight>, KLeft, VOut> get() {
         return new ContextualProcessor<>() {
             private String valueHashSerdePseudoTopic;
-            private Serializer<V> runtimeValueSerializer = constructionTimeValueSerializer;
+            private Serializer<VLeft> runtimeValueSerializer = constructionTimeValueSerializer;
 
-            private KTableValueGetter<K, V> valueGetter;
+            private KTableValueGetter<KLeft, VLeft> valueGetter;
             private Sensor droppedRecordsSensor;
 
 
             @SuppressWarnings({"unchecked", "resource"})
             @Override
-            public void init(final ProcessorContext<K, VR> context) {
+            public void init(final ProcessorContext<KLeft, VOut> context) {
                 super.init(context);
                 valueHashSerdePseudoTopic = valueHashSerdePseudoTopicSupplier.get();
                 valueGetter = valueGetterSupplier.get();
                 valueGetter.init(context);
                 if (runtimeValueSerializer == null) {
-                    runtimeValueSerializer = (Serializer<V>) context.valueSerde().serializer();
+                    runtimeValueSerializer = (Serializer<VLeft>) context.valueSerde().serializer();
                 }
 
                 final InternalProcessorContext<?, ?> internalProcessorContext = (InternalProcessorContext<?, ?>) context;
@@ -98,14 +99,14 @@ public void init(final ProcessorContext<K, VR> context) {
             }
 
             @Override
-            public void process(final Record<K, SubscriptionResponseWrapper<VO>> record) {
+            public void process(final Record<KLeft, SubscriptionResponseWrapper<VRight>> record) {
                 if (record.value().version() != SubscriptionResponseWrapper.CURRENT_VERSION) {
                     //Guard against modifications to SubscriptionResponseWrapper. Need to ensure that there is
                     //compatibility with previous versions to enable rolling upgrades. Must develop a strategy for
                     //upgrading from older SubscriptionWrapper versions to newer versions.
                     throw new UnsupportedVersionException("SubscriptionResponseWrapper is of an incompatible version.");
                 }
-                final ValueAndTimestamp<V> currentValueWithTimestamp = valueGetter.get(record.key());
+                final ValueAndTimestamp<VLeft> currentValueWithTimestamp = valueGetter.get(record.key());
 
                 final long[] currentHash = currentValueWithTimestamp == null ?
                     null :
@@ -115,7 +116,7 @@ public void process(final Record<K, SubscriptionResponseWrapper<VO>> record) {
 
                 //If this value doesn't match the current value from the original table, it is stale and should be discarded.
                 if (java.util.Arrays.equals(messageHash, currentHash)) {
-                    final VR result;
+                    final VOut result;
 
                     if (record.value().foreignValue() == null && (!leftJoin || currentValueWithTimestamp == null)) {
                         result = null; //Emit tombstone
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionJoinProcessorSupplier.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionJoinProcessorSupplier.java
index 008ef665817a1..90dd8c1a365e7 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionJoinProcessorSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionJoinProcessorSupplier.java
@@ -14,7 +14,6 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.apache.kafka.streams.kstream.internals.foreignkeyjoin;
 
 import org.apache.kafka.common.errors.UnsupportedVersionException;
@@ -31,53 +30,44 @@
 import java.util.Objects;
 
 /**
- * Receives {@code SubscriptionWrapper<K>} events and processes them according to their Instruction.
+ * Receives {@code SubscriptionWrapper<KLeft>} events and processes them according to their Instruction.
  * Depending on the results, {@code SubscriptionResponseWrapper}s are created, which will be propagated to
  * the {@code ResponseJoinProcessorSupplier} instance.
  *
- * @param <K> Type of primary keys
- * @param <KO> Type of foreign key
- * @param <VO> Type of foreign value
+ * @param <KLeft> Type of primary keys
+ * @param <KRight> Type of foreign key
+ * @param <VRight> Type of foreign value
  */
-public class SubscriptionJoinProcessorSupplier<K, KO, VO>
-    implements ProcessorSupplier<CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>, K, SubscriptionResponseWrapper<VO>> {
+public class SubscriptionJoinProcessorSupplier<KLeft, KRight, VRight>
+    implements ProcessorSupplier<CombinedKey<KRight, KLeft>, Change<ValueAndTimestamp<SubscriptionWrapper<KLeft>>>, KLeft, SubscriptionResponseWrapper<VRight>> {
 
-    private final KTableValueGetterSupplier<KO, VO> foreignValueGetterSupplier;
+    private final KTableValueGetterSupplier<KRight, VRight> foreignValueGetterSupplier;
 
-    public SubscriptionJoinProcessorSupplier(final KTableValueGetterSupplier<KO, VO> foreignValueGetterSupplier) {
+    public SubscriptionJoinProcessorSupplier(final KTableValueGetterSupplier<KRight, VRight> foreignValueGetterSupplier) {
         this.foreignValueGetterSupplier = foreignValueGetterSupplier;
     }
 
     @Override
-    public Processor<CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>, K, SubscriptionResponseWrapper<VO>> get() {
-
-        return new ContextualProcessor<CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>, K, SubscriptionResponseWrapper<VO>>() {
-
-            private KTableValueGetter<KO, VO> foreignValues;
+    public Processor<CombinedKey<KRight, KLeft>, Change<ValueAndTimestamp<SubscriptionWrapper<KLeft>>>, KLeft, SubscriptionResponseWrapper<VRight>> get() {
+        return new ContextualProcessor<>() {
+            private KTableValueGetter<KRight, VRight> foreignValues;
 
             @Override
-            public void init(final ProcessorContext<K, SubscriptionResponseWrapper<VO>> context) {
+            public void init(final ProcessorContext<KLeft, SubscriptionResponseWrapper<VRight>> context) {
                 super.init(context);
                 foreignValues = foreignValueGetterSupplier.get();
                 foreignValues.init(context);
             }
 
             @Override
-            public void process(final Record<CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>> record) {
+            public void process(final Record<CombinedKey<KRight, KLeft>, Change<ValueAndTimestamp<SubscriptionWrapper<KLeft>>>> record) {
                 Objects.requireNonNull(record.key(), "This processor should never see a null key.");
                 Objects.requireNonNull(record.value(), "This processor should never see a null value.");
-                final ValueAndTimestamp<SubscriptionWrapper<K>> valueAndTimestamp = record.value().newValue;
+                final ValueAndTimestamp<SubscriptionWrapper<KLeft>> valueAndTimestamp = record.value().newValue;
                 Objects.requireNonNull(valueAndTimestamp, "This processor should never see a null newValue.");
-                final SubscriptionWrapper<K> value = valueAndTimestamp.value();
-
-                if (value.version() > SubscriptionWrapper.CURRENT_VERSION) {
-                    //Guard against modifications to SubscriptionWrapper. Need to ensure that there is compatibility
-                    //with previous versions to enable rolling upgrades. Must develop a strategy for upgrading
-                    //from older SubscriptionWrapper versions to newer versions.
-                    throw new UnsupportedVersionException("SubscriptionWrapper is of an incompatible version.");
-                }
+                final SubscriptionWrapper<KLeft> value = subscriptionWrapper(valueAndTimestamp);
 
-                final ValueAndTimestamp<VO> foreignValueAndTime =
+                final ValueAndTimestamp<VRight> foreignValueAndTime =
                     record.key().foreignKey() == null ?
                         null :
                         foreignValues.get(record.key().foreignKey());
@@ -91,7 +81,7 @@ public void process(final Record<CombinedKey<KO, K>, Change<ValueAndTimestamp<Su
                     case DELETE_KEY_AND_PROPAGATE:
                         context().forward(
                             record.withKey(record.key().primaryKey())
-                                .withValue(new SubscriptionResponseWrapper<VO>(
+                                .withValue(new SubscriptionResponseWrapper<VRight>(
                                     value.hash(),
                                     null,
                                     value.primaryPartition()
@@ -103,7 +93,7 @@ public void process(final Record<CombinedKey<KO, K>, Change<ValueAndTimestamp<Su
                         //This one needs to go through regardless of LEFT or INNER join, since the extracted FK was
                         //changed and there is no match for it. We must propagate the (key, null) to ensure that the
                         //downstream consumers are alerted to this fact.
-                        final VO valueToSend = foreignValueAndTime == null ? null : foreignValueAndTime.value();
+                        final VRight valueToSend = foreignValueAndTime == null ? null : foreignValueAndTime.value();
 
                         context().forward(
                             record.withKey(record.key().primaryKey())
@@ -134,6 +124,19 @@ public void process(final Record<CombinedKey<KO, K>, Change<ValueAndTimestamp<Su
                         throw new IllegalStateException("Unhandled instruction: " + value.instruction());
                 }
             }
+
+            private SubscriptionWrapper<KLeft> subscriptionWrapper(final ValueAndTimestamp<SubscriptionWrapper<KLeft>> valueAndTimestamp) {
+                final SubscriptionWrapper<KLeft> value = valueAndTimestamp.value();
+
+                if (value.version() > SubscriptionWrapper.CURRENT_VERSION) {
+                    //Guard against modifications to SubscriptionWrapper. Need to ensure that there is compatibility
+                    //with previous versions to enable rolling upgrades. Must develop a strategy for upgrading
+                    //from older SubscriptionWrapper versions to newer versions.
+                    throw new UnsupportedVersionException("SubscriptionWrapper is of an incompatible version.");
+                }
+
+                return value;
+            }
         };
     }
 }
\ No newline at end of file
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionReceiveProcessorSupplier.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionReceiveProcessorSupplier.java
index 825283e98f581..e654cd752af16 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionReceiveProcessorSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionReceiveProcessorSupplier.java
@@ -14,7 +14,6 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.apache.kafka.streams.kstream.internals.foreignkeyjoin;
 
 import org.apache.kafka.common.errors.UnsupportedVersionException;
@@ -41,16 +40,16 @@
 import java.util.Collections;
 import java.util.Set;
 
-public class SubscriptionReceiveProcessorSupplier<K, KO>
-    implements ProcessorSupplier<KO, SubscriptionWrapper<K>, CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>> {
+public class SubscriptionReceiveProcessorSupplier<KLeft, KRight>
+    implements ProcessorSupplier<KRight, SubscriptionWrapper<KLeft>, CombinedKey<KRight, KLeft>, Change<ValueAndTimestamp<SubscriptionWrapper<KLeft>>>> {
+
     private static final Logger LOG = LoggerFactory.getLogger(SubscriptionReceiveProcessorSupplier.class);
 
     private final StoreFactory subscriptionStoreFactory;
-    private final CombinedKeySchema<KO, K> keySchema;
+    private final CombinedKeySchema<KRight, KLeft> keySchema;
 
     public SubscriptionReceiveProcessorSupplier(final StoreFactory subscriptionStoreFactory,
-                                                final CombinedKeySchema<KO, K> keySchema) {
-
+                                                final CombinedKeySchema<KRight, KLeft> keySchema) {
         this.subscriptionStoreFactory = subscriptionStoreFactory;
         this.keySchema = keySchema;
     }
@@ -61,15 +60,13 @@ public Set<StoreBuilder<?>> stores() {
     }
 
     @Override
-    public Processor<KO, SubscriptionWrapper<K>, CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>> get() {
-
+    public Processor<KRight, SubscriptionWrapper<KLeft>, CombinedKey<KRight, KLeft>, Change<ValueAndTimestamp<SubscriptionWrapper<KLeft>>>> get() {
         return new ContextualProcessor<>() {
-
-            private TimestampedKeyValueStore<Bytes, SubscriptionWrapper<K>> store;
+            private TimestampedKeyValueStore<Bytes, SubscriptionWrapper<KLeft>> store;
             private Sensor droppedRecordsSensor;
 
             @Override
-            public void init(final ProcessorContext<CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>> context) {
+            public void init(final ProcessorContext<CombinedKey<KRight, KLeft>, Change<ValueAndTimestamp<SubscriptionWrapper<KLeft>>>> context) {
                 super.init(context);
                 final InternalProcessorContext<?, ?> internalProcessorContext = (InternalProcessorContext<?, ?>) context;
 
@@ -84,7 +81,7 @@ public void init(final ProcessorContext<CombinedKey<KO, K>, Change<ValueAndTimes
             }
 
             @Override
-            public void process(final Record<KO, SubscriptionWrapper<K>> record) {
+            public void process(final Record<KRight, SubscriptionWrapper<KLeft>> record) {
                 if (record.key() == null && !SubscriptionWrapper.Instruction.PROPAGATE_NULL_IF_NO_FK_VAL_AVAILABLE.equals(record.value().instruction())) {
                     dropRecord();
                     return;
@@ -102,7 +99,7 @@ public void process(final Record<KO, SubscriptionWrapper<K>> record) {
                 );
             }
 
-            private Change<ValueAndTimestamp<SubscriptionWrapper<K>>> inferChange(final Record<KO, SubscriptionWrapper<K>> record) {
+            private Change<ValueAndTimestamp<SubscriptionWrapper<KLeft>>> inferChange(final Record<KRight, SubscriptionWrapper<KLeft>> record) {
                 if (record.key() == null) {
                     return new Change<>(ValueAndTimestamp.make(record.value(), record.timestamp()), null);
                 } else {
@@ -110,11 +107,11 @@ private Change<ValueAndTimestamp<SubscriptionWrapper<K>>> inferChange(final Reco
                 }
             }
 
-            private Change<ValueAndTimestamp<SubscriptionWrapper<K>>> inferBasedOnState(final Record<KO, SubscriptionWrapper<K>> record) {
+            private Change<ValueAndTimestamp<SubscriptionWrapper<KLeft>>> inferBasedOnState(final Record<KRight, SubscriptionWrapper<KLeft>> record) {
                 final Bytes subscriptionKey = keySchema.toBytes(record.key(), record.value().primaryKey());
 
-                final ValueAndTimestamp<SubscriptionWrapper<K>> newValue = ValueAndTimestamp.make(record.value(), record.timestamp());
-                final ValueAndTimestamp<SubscriptionWrapper<K>> oldValue = store.get(subscriptionKey);
+                final ValueAndTimestamp<SubscriptionWrapper<KLeft>> newValue = ValueAndTimestamp.make(record.value(), record.timestamp());
+                final ValueAndTimestamp<SubscriptionWrapper<KLeft>> oldValue = store.get(subscriptionKey);
 
                 //This store is used by the prefix scanner in ForeignTableJoinProcessorSupplier
                 if (record.value().instruction().equals(SubscriptionWrapper.Instruction.DELETE_KEY_AND_PROPAGATE) ||
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionResponseWrapper.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionResponseWrapper.java
index 501edcd243298..e7927e56ea664 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionResponseWrapper.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionResponseWrapper.java
@@ -21,22 +21,22 @@
 import java.util.Arrays;
 import java.util.Objects;
 
-public class SubscriptionResponseWrapper<FV> {
+public class SubscriptionResponseWrapper<VRight> {
     static final byte CURRENT_VERSION = 0;
     // v0 fields:
     private final long[] originalValueHash;
-    private final FV foreignValue;
+    private final VRight foreignValue;
     private final byte version;
     // non-serializing fields
     private final Integer primaryPartition;
 
-    public SubscriptionResponseWrapper(final long[] originalValueHash, final FV foreignValue, final Integer primaryPartition) {
+    public SubscriptionResponseWrapper(final long[] originalValueHash, final VRight foreignValue, final Integer primaryPartition) {
         this(originalValueHash, foreignValue, CURRENT_VERSION, primaryPartition);
     }
 
     public SubscriptionResponseWrapper(
         final long[] originalValueHash,
-        final FV foreignValue,
+        final VRight foreignValue,
         final byte version,
         final Integer primaryPartition) {
         if (version < 0 || version > CURRENT_VERSION) {
@@ -52,7 +52,7 @@ public long[] originalValueHash() {
         return originalValueHash;
     }
 
-    public FV foreignValue() {
+    public VRight foreignValue() {
         return foreignValue;
     }
 
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionResponseWrapperSerde.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionResponseWrapperSerde.java
index 72a918e455f88..0523ed2fe79cf 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionResponseWrapperSerde.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionResponseWrapperSerde.java
@@ -26,22 +26,22 @@
 
 import java.nio.ByteBuffer;
 
-public class SubscriptionResponseWrapperSerde<V> implements Serde<SubscriptionResponseWrapper<V>> {
-    private final SubscriptionResponseWrapperSerializer<V> serializer;
-    private final SubscriptionResponseWrapperDeserializer<V> deserializer;
+public class SubscriptionResponseWrapperSerde<VRight> implements Serde<SubscriptionResponseWrapper<VRight>> {
+    private final SubscriptionResponseWrapperSerializer<VRight> serializer;
+    private final SubscriptionResponseWrapperDeserializer<VRight> deserializer;
 
-    public SubscriptionResponseWrapperSerde(final Serde<V> foreignValueSerde) {
+    public SubscriptionResponseWrapperSerde(final Serde<VRight> foreignValueSerde) {
         serializer = new SubscriptionResponseWrapperSerializer<>(foreignValueSerde == null ? null : foreignValueSerde.serializer());
         deserializer = new SubscriptionResponseWrapperDeserializer<>(foreignValueSerde == null ? null : foreignValueSerde.deserializer());
     }
 
     @Override
-    public Serializer<SubscriptionResponseWrapper<V>> serializer() {
+    public Serializer<SubscriptionResponseWrapper<VRight>> serializer() {
         return serializer;
     }
 
     @Override
-    public Deserializer<SubscriptionResponseWrapper<V>> deserializer() {
+    public Deserializer<SubscriptionResponseWrapper<VRight>> deserializer() {
         return deserializer;
     }
 
@@ -54,7 +54,7 @@ private SubscriptionResponseWrapperSerializer(final Serializer<V> serializer) {
             this.serializer = serializer;
         }
 
-        @SuppressWarnings("unchecked")
+        @SuppressWarnings({"unchecked", "resource"})
         @Override
         public void setIfUnset(final SerdeGetter getter) {
             if (serializer == null) {
@@ -101,7 +101,7 @@ private SubscriptionResponseWrapperDeserializer(final Deserializer<V> deserializ
             this.deserializer = deserializer;
         }
 
-        @SuppressWarnings("unchecked")
+        @SuppressWarnings({"unchecked", "resource"})
         @Override
         public void setIfUnset(final SerdeGetter getter) {
             if (deserializer == null) {
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionSendProcessorSupplier.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionSendProcessorSupplier.java
index 425dbe83bc405..40e404bf301b1 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionSendProcessorSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionSendProcessorSupplier.java
@@ -14,7 +14,6 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.apache.kafka.streams.kstream.internals.foreignkeyjoin;
 
 import org.apache.kafka.common.metrics.Sensor;
@@ -43,22 +42,24 @@
 import static org.apache.kafka.streams.kstream.internals.foreignkeyjoin.SubscriptionWrapper.Instruction.PROPAGATE_NULL_IF_NO_FK_VAL_AVAILABLE;
 import static org.apache.kafka.streams.kstream.internals.foreignkeyjoin.SubscriptionWrapper.Instruction.PROPAGATE_ONLY_IF_FK_VAL_AVAILABLE;
 
-public class SubscriptionSendProcessorSupplier<K, KO, V> implements ProcessorSupplier<K, Change<V>, KO, SubscriptionWrapper<K>> {
+public class SubscriptionSendProcessorSupplier<KLeft, VLeft, KRight>
+    implements ProcessorSupplier<KLeft, Change<VLeft>, KRight, SubscriptionWrapper<KLeft>> {
+
     private static final Logger LOG = LoggerFactory.getLogger(SubscriptionSendProcessorSupplier.class);
 
-    private final ForeignKeyExtractor<? super K, ? super V, ? extends KO> foreignKeyExtractor;
+    private final ForeignKeyExtractor<? super KLeft, ? super VLeft, ? extends KRight> foreignKeyExtractor;
     private final Supplier<String> foreignKeySerdeTopicSupplier;
     private final Supplier<String> valueSerdeTopicSupplier;
     private final boolean leftJoin;
-    private Serializer<KO> foreignKeySerializer;
-    private Serializer<V> valueSerializer;
+    private Serializer<KRight> foreignKeySerializer;
+    private Serializer<VLeft> valueSerializer;
     private boolean useVersionedSemantics;
 
-    public SubscriptionSendProcessorSupplier(final ForeignKeyExtractor<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+    public SubscriptionSendProcessorSupplier(final ForeignKeyExtractor<? super KLeft, ? super VLeft, ? extends KRight> foreignKeyExtractor,
                                              final Supplier<String> foreignKeySerdeTopicSupplier,
                                              final Supplier<String> valueSerdeTopicSupplier,
-                                             final Serde<KO> foreignKeySerde,
-                                             final Serializer<V> valueSerializer,
+                                             final Serde<KRight> foreignKeySerde,
+                                             final Serializer<VLeft> valueSerializer,
                                              final boolean leftJoin) {
         this.foreignKeyExtractor = foreignKeyExtractor;
         this.foreignKeySerdeTopicSupplier = foreignKeySerdeTopicSupplier;
@@ -69,7 +70,7 @@ public SubscriptionSendProcessorSupplier(final ForeignKeyExtractor<? super K, ?
     }
 
     @Override
-    public Processor<K, Change<V>, KO, SubscriptionWrapper<K>> get() {
+    public Processor<KLeft, Change<VLeft>, KRight, SubscriptionWrapper<KLeft>> get() {
         return new UnbindChangeProcessor();
     }
 
@@ -82,7 +83,7 @@ public boolean isUseVersionedSemantics() {
         return useVersionedSemantics;
     }
 
-    private class UnbindChangeProcessor extends ContextualProcessor<K, Change<V>, KO, SubscriptionWrapper<K>> {
+    private class UnbindChangeProcessor extends ContextualProcessor<KLeft, Change<VLeft>, KRight, SubscriptionWrapper<KLeft>> {
 
         private Sensor droppedRecordsSensor;
         private String foreignKeySerdeTopic;
@@ -91,16 +92,16 @@ private class UnbindChangeProcessor extends ContextualProcessor<K, Change<V>, KO
 
         @SuppressWarnings({"unchecked", "resource"})
         @Override
-        public void init(final ProcessorContext<KO, SubscriptionWrapper<K>> context) {
+        public void init(final ProcessorContext<KRight, SubscriptionWrapper<KLeft>> context) {
             super.init(context);
             foreignKeySerdeTopic = foreignKeySerdeTopicSupplier.get();
             valueSerdeTopic = valueSerdeTopicSupplier.get();
             // get default key serde if it wasn't supplied directly at construction
             if (foreignKeySerializer == null) {
-                foreignKeySerializer = (Serializer<KO>) context.keySerde().serializer();
+                foreignKeySerializer = (Serializer<KRight>) context.keySerde().serializer();
             }
             if (valueSerializer == null) {
-                valueSerializer = (Serializer<V>) context.valueSerde().serializer();
+                valueSerializer = (Serializer<VLeft>) context.valueSerde().serializer();
             }
             droppedRecordsSensor = TaskMetrics.droppedRecordsSensor(
                 Thread.currentThread().getName(),
@@ -110,7 +111,7 @@ public void init(final ProcessorContext<KO, SubscriptionWrapper<K>> context) {
         }
 
         @Override
-        public void process(final Record<K, Change<V>> record) {
+        public void process(final Record<KLeft, Change<VLeft>> record) {
             // clear cashed hash from previous record
             recordHash = null;
             // drop out-of-order records from versioned tables (cf. KIP-914)
@@ -126,29 +127,29 @@ public void process(final Record<K, Change<V>> record) {
             }
         }
 
-        private void leftJoinInstructions(final Record<K, Change<V>> record) {
+        private void leftJoinInstructions(final Record<KLeft, Change<VLeft>> record) {
             if (record.value().oldValue != null) {
-                final KO oldForeignKey = foreignKeyExtractor.extract(record.key(), record.value().oldValue);
-                final KO newForeignKey = record.value().newValue == null ? null : foreignKeyExtractor.extract(record.key(), record.value().newValue);
+                final KRight oldForeignKey = foreignKeyExtractor.extract(record.key(), record.value().oldValue);
+                final KRight newForeignKey = record.value().newValue == null ? null : foreignKeyExtractor.extract(record.key(), record.value().newValue);
                 if (oldForeignKey != null && !Arrays.equals(serialize(newForeignKey), serialize(oldForeignKey))) {
                     forward(record, oldForeignKey, DELETE_KEY_AND_PROPAGATE);
                 }
                 forward(record, newForeignKey, PROPAGATE_NULL_IF_NO_FK_VAL_AVAILABLE);
             } else if (record.value().newValue != null) {
-                final KO newForeignKey = foreignKeyExtractor.extract(record.key(), record.value().newValue);
+                final KRight newForeignKey = foreignKeyExtractor.extract(record.key(), record.value().newValue);
                 forward(record, newForeignKey, PROPAGATE_NULL_IF_NO_FK_VAL_AVAILABLE);
             }
         }
 
-        private void defaultJoinInstructions(final Record<K, Change<V>> record) {
+        private void defaultJoinInstructions(final Record<KLeft, Change<VLeft>> record) {
             if (record.value().oldValue != null) {
-                final KO oldForeignKey = record.value().oldValue == null ? null : foreignKeyExtractor.extract(record.key(), record.value().oldValue);
+                final KRight oldForeignKey = record.value().oldValue == null ? null : foreignKeyExtractor.extract(record.key(), record.value().oldValue);
                 if (oldForeignKey == null) {
                     logSkippedRecordDueToNullForeignKey();
                     return;
                 }
                 if (record.value().newValue != null) {
-                    final KO newForeignKey = record.value().newValue == null ? null : foreignKeyExtractor.extract(record.key(), record.value().newValue);
+                    final KRight newForeignKey = record.value().newValue == null ? null : foreignKeyExtractor.extract(record.key(), record.value().newValue);
                     if (newForeignKey == null) {
                         logSkippedRecordDueToNullForeignKey();
                         return;
@@ -166,7 +167,7 @@ private void defaultJoinInstructions(final Record<K, Change<V>> record) {
                     forward(record, oldForeignKey, DELETE_KEY_AND_PROPAGATE);
                 }
             } else if (record.value().newValue != null) {
-                final KO newForeignKey = foreignKeyExtractor.extract(record.key(), record.value().newValue);
+                final KRight newForeignKey = foreignKeyExtractor.extract(record.key(), record.value().newValue);
                 if (newForeignKey == null) {
                     logSkippedRecordDueToNullForeignKey();
                 } else {
@@ -175,12 +176,12 @@ private void defaultJoinInstructions(final Record<K, Change<V>> record) {
             }
         }
 
-        private byte[] serialize(final KO key) {
+        private byte[] serialize(final KRight key) {
             return foreignKeySerializer.serialize(foreignKeySerdeTopic, key);
         }
 
-        private void forward(final Record<K, Change<V>> record, final KO foreignKey, final Instruction deleteKeyNoPropagate) {
-            final SubscriptionWrapper<K> wrapper = new SubscriptionWrapper<>(
+        private void forward(final Record<KLeft, Change<VLeft>> record, final KRight foreignKey, final Instruction deleteKeyNoPropagate) {
+            final SubscriptionWrapper<KLeft> wrapper = new SubscriptionWrapper<>(
                 hash(record),
                 deleteKeyNoPropagate,
                 record.key(),
@@ -189,7 +190,7 @@ private void forward(final Record<K, Change<V>> record, final KO foreignKey, fin
             context().forward(record.withKey(foreignKey).withValue(wrapper));
         }
 
-        private long[] hash(final Record<K, Change<V>> record) {
+        private long[] hash(final Record<KLeft, Change<VLeft>> record) {
             if (recordHash == null) {
                 recordHash = record.value().newValue == null
                     ? null
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionWrapper.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionWrapper.java
index b104d420c704b..0a14eedcb6eab 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionWrapper.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionWrapper.java
@@ -21,8 +21,7 @@
 import java.util.Arrays;
 import java.util.Objects;
 
-
-public class SubscriptionWrapper<K> {
+public class SubscriptionWrapper<KLeft> {
     static final byte VERSION_0 = 0;
     static final byte VERSION_1 = 1;
 
@@ -32,7 +31,7 @@ public class SubscriptionWrapper<K> {
     private final long[] hash;
     private final Instruction instruction;
     private final byte version;
-    private final K primaryKey;
+    private final KLeft primaryKey;
     // v1 fields:
     private final Integer primaryPartition;
 
@@ -71,11 +70,11 @@ public static Instruction fromValue(final byte value) {
         }
     }
 
-    public SubscriptionWrapper(final long[] hash, final Instruction instruction, final K primaryKey, final Integer primaryPartition) {
+    public SubscriptionWrapper(final long[] hash, final Instruction instruction, final KLeft primaryKey, final Integer primaryPartition) {
         this(hash, instruction, primaryKey, CURRENT_VERSION, primaryPartition);
     }
 
-    public SubscriptionWrapper(final long[] hash, final Instruction instruction, final K primaryKey, final byte version, final Integer primaryPartition) {
+    public SubscriptionWrapper(final long[] hash, final Instruction instruction, final KLeft primaryKey, final byte version, final Integer primaryPartition) {
         Objects.requireNonNull(instruction, "instruction cannot be null. Required by downstream processor.");
         Objects.requireNonNull(primaryKey, "primaryKey cannot be null. Required by downstream processor.");
         if (version < 0 || version > CURRENT_VERSION) {
@@ -97,7 +96,7 @@ public long[] hash() {
         return hash;
     }
 
-    public K primaryKey() {
+    public KLeft primaryKey() {
         return primaryKey;
     }
 
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionWrapperSerde.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionWrapperSerde.java
index ed639eba0fe25..b03b24749e0b9 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionWrapperSerde.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionWrapperSerde.java
@@ -31,9 +31,10 @@
 import java.util.Map;
 import java.util.function.Supplier;
 
-public class SubscriptionWrapperSerde<K> extends WrappingNullableSerde<SubscriptionWrapper<K>, K, Void> {
+public class SubscriptionWrapperSerde<KLeft> extends WrappingNullableSerde<SubscriptionWrapper<KLeft>, KLeft, Void> {
+
     public SubscriptionWrapperSerde(final Supplier<String> primaryKeySerializationPseudoTopicSupplier,
-                                    final Serde<K> primaryKeySerde) {
+                                    final Serde<KLeft> primaryKeySerde) {
         super(
             new SubscriptionWrapperSerializer<>(primaryKeySerializationPseudoTopicSupplier,
                                                 primaryKeySerde == null ? null : primaryKeySerde.serializer()),
@@ -42,25 +43,25 @@ public SubscriptionWrapperSerde(final Supplier<String> primaryKeySerializationPs
         );
     }
 
-    private static class SubscriptionWrapperSerializer<K>
-        implements Serializer<SubscriptionWrapper<K>>, WrappingNullableSerializer<SubscriptionWrapper<K>, K, Void> {
+    private static class SubscriptionWrapperSerializer<KLeft>
+        implements Serializer<SubscriptionWrapper<KLeft>>, WrappingNullableSerializer<SubscriptionWrapper<KLeft>, KLeft, Void> {
 
         private final Supplier<String> primaryKeySerializationPseudoTopicSupplier;
         private String primaryKeySerializationPseudoTopic = null;
-        private Serializer<K> primaryKeySerializer;
+        private Serializer<KLeft> primaryKeySerializer;
         private boolean upgradeFromV0 = false;
 
         SubscriptionWrapperSerializer(final Supplier<String> primaryKeySerializationPseudoTopicSupplier,
-                                      final Serializer<K> primaryKeySerializer) {
+                                      final Serializer<KLeft> primaryKeySerializer) {
             this.primaryKeySerializationPseudoTopicSupplier = primaryKeySerializationPseudoTopicSupplier;
             this.primaryKeySerializer = primaryKeySerializer;
         }
 
-        @SuppressWarnings("unchecked")
+        @SuppressWarnings({"unchecked", "resource"})
         @Override
         public void setIfUnset(final SerdeGetter getter) {
             if (primaryKeySerializer == null) {
-                primaryKeySerializer = (Serializer<K>) getter.keySerde().serializer();
+                primaryKeySerializer = (Serializer<KLeft>) getter.keySerde().serializer();
             }
         }
 
@@ -103,12 +104,11 @@ private static boolean upgradeFromV0(final Map<String, ?> configs) {
         }
 
         @Override
-        public byte[] serialize(final String ignored, final SubscriptionWrapper<K> data) {
+        public byte[] serialize(final String ignored, final SubscriptionWrapper<KLeft> data) {
             //{1-bit-isHashNull}{7-bits-version}{1-byte-instruction}{Optional-16-byte-Hash}{PK-serialized}{4-bytes-primaryPartition}
 
-            //7-bit (0x7F) maximum for data version.
-            if (Byte.compare((byte) 0x7F, data.version()) < 0) {
-                throw new UnsupportedVersionException("SubscriptionWrapper version is larger than maximum supported 0x7F");
+            if (data.version() < 0) {
+                throw new UnsupportedVersionException("SubscriptionWrapper version cannot be negative");
             }
 
             final int version = data.version();
@@ -121,7 +121,7 @@ public byte[] serialize(final String ignored, final SubscriptionWrapper<K> data)
             }
         }
 
-        private byte[] serializePrimaryKey(final SubscriptionWrapper<K> data) {
+        private byte[] serializePrimaryKey(final SubscriptionWrapper<KLeft> data) {
             if (primaryKeySerializationPseudoTopic == null) {
                 primaryKeySerializationPseudoTopic = primaryKeySerializationPseudoTopicSupplier.get();
             }
@@ -132,7 +132,7 @@ private byte[] serializePrimaryKey(final SubscriptionWrapper<K> data) {
             );
         }
 
-        private ByteBuffer serializeCommon(final SubscriptionWrapper<K> data, final byte version, final int extraLength) {
+        private ByteBuffer serializeCommon(final SubscriptionWrapper<KLeft> data, final byte version, final int extraLength) {
             final byte[] primaryKeySerializedData = serializePrimaryKey(data);
             final ByteBuffer buf;
             int dataLength = 2 + primaryKeySerializedData.length + extraLength;
@@ -155,40 +155,40 @@ private ByteBuffer serializeCommon(final SubscriptionWrapper<K> data, final byte
             return buf;
         }
 
-        private byte[] serializeV0(final SubscriptionWrapper<K> data) {
+        private byte[] serializeV0(final SubscriptionWrapper<KLeft> data) {
             return serializeCommon(data, (byte) 0, 0).array();
         }
 
-        private byte[] serializeV1(final SubscriptionWrapper<K> data) {
+        private byte[] serializeV1(final SubscriptionWrapper<KLeft> data) {
             final ByteBuffer buf = serializeCommon(data, data.version(), Integer.BYTES);
             buf.putInt(data.primaryPartition());
             return buf.array();
         }
     }
 
-    private static class SubscriptionWrapperDeserializer<K>
-        implements Deserializer<SubscriptionWrapper<K>>, WrappingNullableDeserializer<SubscriptionWrapper<K>, K, Void> {
+    private static class SubscriptionWrapperDeserializer<KLeft>
+        implements Deserializer<SubscriptionWrapper<KLeft>>, WrappingNullableDeserializer<SubscriptionWrapper<KLeft>, KLeft, Void> {
 
         private final Supplier<String> primaryKeySerializationPseudoTopicSupplier;
         private String primaryKeySerializationPseudoTopic = null;
-        private Deserializer<K> primaryKeyDeserializer;
+        private Deserializer<KLeft> primaryKeyDeserializer;
 
         SubscriptionWrapperDeserializer(final Supplier<String> primaryKeySerializationPseudoTopicSupplier,
-                                        final Deserializer<K> primaryKeyDeserializer) {
+                                        final Deserializer<KLeft> primaryKeyDeserializer) {
             this.primaryKeySerializationPseudoTopicSupplier = primaryKeySerializationPseudoTopicSupplier;
             this.primaryKeyDeserializer = primaryKeyDeserializer;
         }
 
-        @SuppressWarnings("unchecked")
+        @SuppressWarnings({"unchecked", "resource"})
         @Override
         public void setIfUnset(final SerdeGetter getter) {
             if (primaryKeyDeserializer == null) {
-                primaryKeyDeserializer = (Deserializer<K>) getter.keySerde().deserializer();
+                primaryKeyDeserializer = (Deserializer<KLeft>) getter.keySerde().deserializer();
             }
         }
 
         @Override
-        public SubscriptionWrapper<K> deserialize(final String ignored, final byte[] data) {
+        public SubscriptionWrapper<KLeft> deserialize(final String ignored, final byte[] data) {
             //{7-bits-version}{1-bit-isHashNull}{1-byte-instruction}{Optional-16-byte-Hash}{PK-serialized}{4-bytes-primaryPartition}
             final ByteBuffer buf = ByteBuffer.wrap(data);
             final byte versionAndIsHashNull = buf.get();
@@ -220,7 +220,7 @@ public SubscriptionWrapper<K> deserialize(final String ignored, final byte[] dat
                 primaryKeySerializationPseudoTopic = primaryKeySerializationPseudoTopicSupplier.get();
             }
 
-            final K primaryKey = primaryKeyDeserializer.deserialize(
+            final KLeft primaryKey = primaryKeyDeserializer.deserialize(
                 primaryKeySerializationPseudoTopic,
                 primaryKeyRaw
             );
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionSendProcessorSupplierTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionSendProcessorSupplierTest.java
index 87366bd5334a9..fe262840a432e 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionSendProcessorSupplierTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionSendProcessorSupplierTest.java
@@ -44,7 +44,7 @@
 public class SubscriptionSendProcessorSupplierTest {
 
     private final Processor<String, Change<LeftValue>, String, SubscriptionWrapper<String>> leftJoinProcessor =
-        new SubscriptionSendProcessorSupplier<String, String, LeftValue>(
+        new SubscriptionSendProcessorSupplier<String, LeftValue, String>(
             ForeignKeyExtractor.fromFunction(LeftValue::getForeignKey),
             () -> "subscription-topic-fk",
             () -> "value-serde-topic",
@@ -54,7 +54,7 @@ public class SubscriptionSendProcessorSupplierTest {
         ).get();
 
     private final Processor<String, Change<LeftValue>, String, SubscriptionWrapper<String>> innerJoinProcessor =
-        new SubscriptionSendProcessorSupplier<String, String, LeftValue>(
+        new SubscriptionSendProcessorSupplier<String, LeftValue, String>(
             ForeignKeyExtractor.fromFunction(LeftValue::getForeignKey),
             () -> "subscription-topic-fk",
             () -> "value-serde-topic",
@@ -329,7 +329,7 @@ public void innerJoinShouldPropagateNothingWhenOldAndNewLeftValueIsNull() {
 
     // Bi-function tests: inner join, left join
     private final Processor<String, Change<LeftValue>, String, SubscriptionWrapper<String>> biFunctionLeftJoinProcessor =
-        new SubscriptionSendProcessorSupplier<String, String, LeftValue>(
+        new SubscriptionSendProcessorSupplier<String, LeftValue, String>(
             ForeignKeyExtractor.fromBiFunction((key, value) -> value.getForeignKey() == null ? null : key + value.getForeignKey()),
             () -> "subscription-topic-fk",
             () -> "value-serde-topic",
@@ -339,7 +339,7 @@ public void innerJoinShouldPropagateNothingWhenOldAndNewLeftValueIsNull() {
         ).get();
 
     private final Processor<String, Change<LeftValue>, String, SubscriptionWrapper<String>> biFunctionInnerJoinProcessor =
-        new SubscriptionSendProcessorSupplier<String, String, LeftValue>(
+        new SubscriptionSendProcessorSupplier<String, LeftValue, String>(
             ForeignKeyExtractor.fromBiFunction((key, value) -> value.getForeignKey() == null ? null : key + value.getForeignKey()),
             () -> "subscription-topic-fk",
             () -> "value-serde-topic",
@@ -628,6 +628,7 @@ public void biFunctionInnerJoinShouldPropagateNothingWhenOldAndNewLeftValueIsNul
     }
 
     private static class LeftValueSerializer implements Serializer<LeftValue> {
+        @SuppressWarnings("resource")
         @Override
         public byte[] serialize(final String topic, final LeftValue data) {
             if (data == null) return null;
@@ -648,6 +649,7 @@ public String getForeignKey() {
         }
     }
 
+    @SuppressWarnings("resource")
     private static long[] hash(final LeftValue value) {
         return Murmur3.hash128(new LeftValueSerializer().serialize("value-serde-topic", value));
     }