diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 13214c14..d0eb7aae 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -2,7 +2,7 @@
 junit = "5.13.3"
 mockito = "5.18.0"
 testcontainers = "1.21.3"
-kafkaUtils = "1.2.1"
+kafkaUtils = "1.2.2-SNAPSHOT"
 
 [libraries]
 kafka-bom = { group = "com.bakdata.kafka", name = "kafka-bom", version.ref = "kafkaUtils" }
@@ -17,7 +17,7 @@ kafka-streams-avro-serde = { group = "io.confluent", name = "kafka-streams-avro-
 kafka-protobuf-provider = { group = "io.confluent", name = "kafka-protobuf-provider" }
 largeMessage-bom = { group = "com.bakdata.kafka", name = "large-message-bom", version = "3.1.0" }
 largeMessage-core = { group = "com.bakdata.kafka", name = "large-message-core" }
-errorHandling-bom = { group = "com.bakdata.kafka", name = "error-handling-bom", version = "2.0.0" }
+errorHandling-bom = { group = "com.bakdata.kafka", name = "error-handling-bom", version = "2.0.1-SNAPSHOT" }
 errorHandling-core = { group = "com.bakdata.kafka", name = "error-handling-core" }
 picocli = { group = "info.picocli", name = "picocli", version = "4.7.7" }
 slf4j = { group = "org.slf4j", name = "slf4j-api", version = "2.0.17" }
@@ -33,7 +33,7 @@ mockito-core = { group = "org.mockito", name = "mockito-core", version.ref = "mo
 mockito-junit = { group = "org.mockito", name = "mockito-junit-jupiter", version.ref = "mockito" }
 testcontainers-junit = { group = "org.testcontainers", name = "junit-jupiter", version.ref = "testcontainers" }
 testcontainers-kafka = { group = "org.testcontainers", name = "kafka", version.ref = "testcontainers" }
-fluentKafkaStreamsTests = { group = "com.bakdata.fluent-kafka-streams-tests", name = "fluent-kafka-streams-tests-junit5", version = "3.4.1" }
+fluentKafkaStreamsTests = { group = "com.bakdata.fluent-kafka-streams-tests", name = "fluent-kafka-streams-tests-junit5", version = "3.4.2-SNAPSHOT" }
 log4j-slf4j2 = { group = "org.apache.logging.log4j", name = "log4j-slf4j2-impl", version = "2.25.1" }
 awaitility = { group = "org.awaitility", name = "awaitility", version = "4.3.0" }
 
diff --git a/streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KStreamX.java b/streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KStreamX.java
index 55f3b016..668052f2 100644
--- a/streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KStreamX.java
+++ b/streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KStreamX.java
@@ -850,7 +850,7 @@ <GK, GV, RV> KStreamX<K, RV> leftJoin(GlobalKTable<GK, GV> globalTable,
 
     @Override
     <KOut, VOut> KStreamX<KOut, VOut> process(
-            ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier,
+            ProcessorSupplier<? super K, ? super V, ? extends KOut, ? extends VOut> processorSupplier,
             String... stateStoreNames);
 
     /**
@@ -866,7 +866,7 @@ <KOut, VOut> KStreamX<KOut, VOut> process(
      * @see ErrorCapturingProcessor#captureErrors(ProcessorSupplier)
      */
     <KOut, VOut> KErrorStreamX<K, V, KOut, VOut> processCapturingErrors(
-            ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier,
+            ProcessorSupplier<? super K, ? super V, ? extends KOut, ? extends VOut> processorSupplier,
             String... stateStoreNames);
 
     /**
@@ -883,13 +883,13 @@ <KOut, VOut> KErrorStreamX<K, V, KOut, VOut> processCapturingErrors(
      * @see ErrorCapturingProcessor#captureErrors(ProcessorSupplier, java.util.function.Predicate)
      */
     <KOut, VOut> KErrorStreamX<K, V, KOut, VOut> processCapturingErrors(
-            ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier,
+            ProcessorSupplier<? super K, ? super V, ? extends KOut, ? extends VOut> processorSupplier,
             java.util.function.Predicate<Exception> errorFilter,
             String... stateStoreNames);
 
     @Override
     <KOut, VOut> KStreamX<KOut, VOut> process(
-            ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier,
+            ProcessorSupplier<? super K, ? super V, ? extends KOut, ? extends VOut> processorSupplier,
             Named named, String... stateStoreNames);
 
     /**
@@ -906,7 +906,7 @@ <KOut, VOut> KStreamX<KOut, VOut> process(
      * @see ErrorCapturingProcessor#captureErrors(ProcessorSupplier)
      */
     <KOut, VOut> KErrorStreamX<K, V, KOut, VOut> processCapturingErrors(
-            ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier,
+            ProcessorSupplier<? super K, ? super V, ? extends KOut, ? extends VOut> processorSupplier,
             Named named, String... stateStoreNames);
 
     /**
@@ -924,13 +924,13 @@ <KOut, VOut> KErrorStreamX<K, V, KOut, VOut> processCapturingErrors(
      * @see ErrorCapturingProcessor#captureErrors(ProcessorSupplier, java.util.function.Predicate)
      */
     <KOut, VOut> KErrorStreamX<K, V, KOut, VOut> processCapturingErrors(
-            ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier,
+            ProcessorSupplier<? super K, ? super V, ? extends KOut, ? extends VOut> processorSupplier,
             java.util.function.Predicate<Exception> errorFilter,
             Named named, String... stateStoreNames);
 
     @Override
     <VOut> KStreamX<K, VOut> processValues(
-            FixedKeyProcessorSupplier<? super K, ? super V, VOut> processorSupplier,
+            FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
             String... stateStoreNames);
 
     /**
@@ -947,7 +947,7 @@ <VOut> KStreamX<K, VOut> processValues(
      * @see ErrorCapturingValueProcessor#captureErrors(FixedKeyProcessorSupplier)
      */
     <VOut> KErrorStreamX<K, V, K, VOut> processValuesCapturingErrors(
-            FixedKeyProcessorSupplier<? super K, ? super V, VOut> processorSupplier,
+            FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
             String... stateStoreNames);
 
     /**
@@ -965,13 +965,13 @@ <VOut> KErrorStreamX<K, V, K, VOut> processValuesCapturingErrors(
      * @see ErrorCapturingValueProcessor#captureErrors(FixedKeyProcessorSupplier, java.util.function.Predicate)
      */
     <VOut> KErrorStreamX<K, V, K, VOut> processValuesCapturingErrors(
-            FixedKeyProcessorSupplier<? super K, ? super V, VOut> processorSupplier,
+            FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
             java.util.function.Predicate<Exception> errorFilter,
             String... stateStoreNames);
 
     @Override
     <VOut> KStreamX<K, VOut> processValues(
-            FixedKeyProcessorSupplier<? super K, ? super V, VOut> processorSupplier,
+            FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
             Named named, String... stateStoreNames);
 
     /**
@@ -989,7 +989,7 @@ <VOut> KStreamX<K, VOut> processValues(
      * @see ErrorCapturingValueProcessor#captureErrors(FixedKeyProcessorSupplier)
      */
     <VOut> KErrorStreamX<K, V, K, VOut> processValuesCapturingErrors(
-            FixedKeyProcessorSupplier<? super K, ? super V, VOut> processorSupplier,
+            FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
             Named named, String... stateStoreNames);
 
     /**
@@ -1008,7 +1008,7 @@ <VOut> KErrorStreamX<K, V, K, VOut> processValuesCapturingErrors(
      * @see ErrorCapturingValueProcessor#captureErrors(FixedKeyProcessorSupplier, java.util.function.Predicate)
      */
     <VOut> KErrorStreamX<K, V, K, VOut> processValuesCapturingErrors(
-            FixedKeyProcessorSupplier<? super K, ? super V, VOut> processorSupplier,
+            FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
             java.util.function.Predicate<Exception> errorFilter,
             Named named, String... stateStoreNames);
 }
diff --git a/streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KStreamXImpl.java b/streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KStreamXImpl.java
index 0bdd0b6a..876c643f 100644
--- a/streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KStreamXImpl.java
+++ b/streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KStreamXImpl.java
@@ -838,14 +838,14 @@ public <GK, GV, RV> KStreamX<K, RV> leftJoin(final GlobalKTable<GK, GV> globalTa
 
     @Override
     public <KOut, VOut> KStreamX<KOut, VOut> process(
-            final ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier,
+            final ProcessorSupplier<? super K, ? super V, ? extends KOut, ? extends VOut> processorSupplier,
             final String... stateStoreNames) {
         return this.context.wrap(this.wrapped.process(processorSupplier, stateStoreNames));
     }
 
     @Override
     public <KOut, VOut> KErrorStreamX<K, V, KOut, VOut> processCapturingErrors(
-            final ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier,
+            final ProcessorSupplier<? super K, ? super V, ? extends KOut, ? extends VOut> processorSupplier,
             final String... stateStoreNames) {
         return this.processCapturingErrorsInternal(ErrorCapturingProcessor.captureErrors(processorSupplier),
                 stateStoreNames);
@@ -853,7 +853,7 @@ public <KOut, VOut> KErrorStreamX<K, V, KOut, VOut> processCapturingErrors(
 
     @Override
     public <KOut, VOut> KErrorStreamX<K, V, KOut, VOut> processCapturingErrors(
-            final ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier,
+            final ProcessorSupplier<? super K, ? super V, ? extends KOut, ? extends VOut> processorSupplier,
             final java.util.function.Predicate<Exception> errorFilter,
             final String... stateStoreNames) {
         return this.processCapturingErrorsInternal(
@@ -863,14 +863,16 @@ public <KOut, VOut> KErrorStreamX<K, V, KOut, VOut> processCapturingErrors(
 
     @Override
     public <KOut, VOut> KStreamX<KOut, VOut> process(
-            final ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier, final Named named,
+            final ProcessorSupplier<? super K, ? super V, ? extends KOut, ? extends VOut> processorSupplier,
+            final Named named,
             final String... stateStoreNames) {
         return this.context.wrap(this.wrapped.process(processorSupplier, named, stateStoreNames));
     }
 
     @Override
     public <KOut, VOut> KErrorStreamX<K, V, KOut, VOut> processCapturingErrors(
-            final ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier, final Named named,
+            final ProcessorSupplier<? super K, ? super V, ? extends KOut, ? extends VOut> processorSupplier,
+            final Named named,
             final String... stateStoreNames) {
         return this.processCapturingErrorsInternal(ErrorCapturingProcessor.captureErrors(processorSupplier), named,
                 stateStoreNames);
@@ -878,7 +880,7 @@ public <KOut, VOut> KErrorStreamX<K, V, KOut, VOut> processCapturingErrors(
 
     @Override
     public <KOut, VOut> KErrorStreamX<K, V, KOut, VOut> processCapturingErrors(
-            final ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier,
+            final ProcessorSupplier<? super K, ? super V, ? extends KOut, ? extends VOut> processorSupplier,
             final java.util.function.Predicate<Exception> errorFilter,
             final Named named, final String... stateStoreNames) {
         return this.processCapturingErrorsInternal(
@@ -888,14 +890,14 @@ public <KOut, VOut> KErrorStreamX<K, V, KOut, VOut> processCapturingErrors(
 
     @Override
     public <VOut> KStreamX<K, VOut> processValues(
-            final FixedKeyProcessorSupplier<? super K, ? super V, VOut> processorSupplier,
+            final FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
             final String... stateStoreNames) {
         return this.context.wrap(this.wrapped.processValues(processorSupplier, stateStoreNames));
     }
 
     @Override
     public <VOut> KErrorStreamX<K, V, K, VOut> processValuesCapturingErrors(
-            final FixedKeyProcessorSupplier<? super K, ? super V, VOut> processorSupplier,
+            final FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
             final String... stateStoreNames) {
         return this.processValuesCapturingErrorsInternal(ErrorCapturingValueProcessor.captureErrors(processorSupplier),
                 stateStoreNames);
@@ -903,7 +905,7 @@ public <VOut> KErrorStreamX<K, V, K, VOut> processValuesCapturingErrors(
 
     @Override
     public <VOut> KErrorStreamX<K, V, K, VOut> processValuesCapturingErrors(
-            final FixedKeyProcessorSupplier<? super K, ? super V, VOut> processorSupplier,
+            final FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
             final java.util.function.Predicate<Exception> errorFilter, final String... stateStoreNames) {
         return this.processValuesCapturingErrorsInternal(
                 ErrorCapturingValueProcessor.captureErrors(processorSupplier, errorFilter),
@@ -913,14 +915,14 @@ public <VOut> KErrorStreamX<K, V, K, VOut> processValuesCapturingErrors(
 
     @Override
     public <VOut> KStreamX<K, VOut> processValues(
-            final FixedKeyProcessorSupplier<? super K, ? super V, VOut> processorSupplier, final Named named,
+            final FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier, final Named named,
             final String... stateStoreNames) {
         return this.context.wrap(this.wrapped.processValues(processorSupplier, named, stateStoreNames));
     }
 
     @Override
     public <VOut> KErrorStreamX<K, V, K, VOut> processValuesCapturingErrors(
-            final FixedKeyProcessorSupplier<? super K, ? super V, VOut> processorSupplier, final Named named,
+            final FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier, final Named named,
             final String... stateStoreNames) {
         return this.processValuesCapturingErrorsInternal(ErrorCapturingValueProcessor.captureErrors(processorSupplier),
                 named,
@@ -929,87 +931,87 @@ public <VOut> KErrorStreamX<K, V, K, VOut> processValuesCapturingErrors(
 
     @Override
     public <VOut> KErrorStreamX<K, V, K, VOut> processValuesCapturingErrors(
-            final FixedKeyProcessorSupplier<? super K, ? super V, VOut> processorSupplier,
+            final FixedKeyProcessorSupplier<? super K, ? super V, ? extends VOut> processorSupplier,
             final java.util.function.Predicate<Exception> errorFilter, final Named named,
             final String... stateStoreNames) {
         return this.processValuesCapturingErrorsInternal(
                 ErrorCapturingValueProcessor.captureErrors(processorSupplier, errorFilter), named, stateStoreNames);
     }
 
-    private <KR, VR> KeyValueKErrorStreamX<K, V, KR, VR> mapCapturingErrorsInternal(
+    private <KR, VR> KErrorStreamX<K, V, KR, VR> mapCapturingErrorsInternal(
             final KeyValueMapper<K, V, KeyValue<KR, ProcessedKeyValue<K, V, VR>>> mapper) {
         final KStreamX<KR, ProcessedKeyValue<K, V, VR>> map = this.map(mapper);
         return new KeyValueKErrorStreamX<>(map);
     }
 
-    private <KR, VR> KeyValueKErrorStreamX<K, V, KR, VR> mapCapturingErrorsInternal(
+    private <KR, VR> KErrorStreamX<K, V, KR, VR> mapCapturingErrorsInternal(
             final KeyValueMapper<K, V, KeyValue<KR, ProcessedKeyValue<K, V, VR>>> mapper, final Named named) {
         final KStreamX<KR, ProcessedKeyValue<K, V, VR>> map = this.map(mapper, named);
         return new KeyValueKErrorStreamX<>(map);
     }
 
-    private <VR> ValueKErrorStreamX<K, V, VR> mapValuesCapturingErrorsInternal(
+    private <VR> KErrorStreamX<K, V, K, VR> mapValuesCapturingErrorsInternal(
             final ValueMapper<V, ProcessedValue<V, VR>> mapper) {
         final KStreamX<K, ProcessedValue<V, VR>> map = this.mapValues(mapper);
         return new ValueKErrorStreamX<>(map);
     }
 
-    private <VR> ValueKErrorStreamX<K, V, VR> mapValuesCapturingErrorsInternal(
+    private <VR> KErrorStreamX<K, V, K, VR> mapValuesCapturingErrorsInternal(
             final ValueMapper<V, ProcessedValue<V, VR>> mapper,
             final Named named) {
         final KStreamX<K, ProcessedValue<V, VR>> map = this.mapValues(mapper, named);
         return new ValueKErrorStreamX<>(map);
     }
 
-    private <VR> ValueKErrorStreamX<K, V, VR> mapValuesCapturingErrorsInternal(
+    private <VR> KErrorStreamX<K, V, K, VR> mapValuesCapturingErrorsInternal(
             final ValueMapperWithKey<K, V, ProcessedValue<V, VR>> mapper) {
         final KStreamX<K, ProcessedValue<V, VR>> map = this.mapValues(mapper);
         return new ValueKErrorStreamX<>(map);
     }
 
-    private <VR> ValueKErrorStreamX<K, V, VR> mapValuesCapturingErrorsInternal(
+    private <VR> KErrorStreamX<K, V, K, VR> mapValuesCapturingErrorsInternal(
             final ValueMapperWithKey<K, V, ProcessedValue<V, VR>> mapper, final Named named) {
         final KStreamX<K, ProcessedValue<V, VR>> map = this.mapValues(mapper, named);
         return new ValueKErrorStreamX<>(map);
     }
 
-    private <KR, VR> KeyValueKErrorStreamX<K, V, KR, VR> flatMapCapturingErrorsInternal(
+    private <KR, VR> KErrorStreamX<K, V, KR, VR> flatMapCapturingErrorsInternal(
             final KeyValueMapper<K, V, Iterable<KeyValue<KR, ProcessedKeyValue<K, V, VR>>>> mapper) {
         final KStreamX<KR, ProcessedKeyValue<K, V, VR>> map = this.flatMap(mapper);
         return new KeyValueKErrorStreamX<>(map);
     }
 
-    private <KR, VR> KeyValueKErrorStreamX<K, V, KR, VR> flatMapCapturingErrorsInternal(
+    private <KR, VR> KErrorStreamX<K, V, KR, VR> flatMapCapturingErrorsInternal(
             final KeyValueMapper<K, V, Iterable<KeyValue<KR, ProcessedKeyValue<K, V, VR>>>> mapper, final Named named) {
         final KStreamX<KR, ProcessedKeyValue<K, V, VR>> map = this.flatMap(mapper, named);
         return new KeyValueKErrorStreamX<>(map);
     }
 
-    private <VR> ValueKErrorStreamX<K, V, VR> flatMapValuesCapturingErrorsInternal(
+    private <VR> KErrorStreamX<K, V, K, VR> flatMapValuesCapturingErrorsInternal(
             final ValueMapper<V, Iterable<ProcessedValue<V, VR>>> mapper) {
         final KStreamX<K, ProcessedValue<V, VR>> map = this.flatMapValues(mapper);
         return new ValueKErrorStreamX<>(map);
     }
 
-    private <VR> ValueKErrorStreamX<K, V, VR> flatMapValuesCapturingErrorsInternal(
+    private <VR> KErrorStreamX<K, V, K, VR> flatMapValuesCapturingErrorsInternal(
             final ValueMapper<V, Iterable<ProcessedValue<V, VR>>> mapper, final Named named) {
         final KStreamX<K, ProcessedValue<V, VR>> map = this.flatMapValues(mapper, named);
         return new ValueKErrorStreamX<>(map);
     }
 
-    private <VR> ValueKErrorStreamX<K, V, VR> flatMapValuesCapturingErrorsInternal(
+    private <VR> KErrorStreamX<K, V, K, VR> flatMapValuesCapturingErrorsInternal(
             final ValueMapperWithKey<K, V, Iterable<ProcessedValue<V, VR>>> mapper) {
         final KStreamX<K, ProcessedValue<V, VR>> map = this.flatMapValues(mapper);
         return new ValueKErrorStreamX<>(map);
     }
 
-    private <VR> ValueKErrorStreamX<K, V, VR> flatMapValuesCapturingErrorsInternal(
+    private <VR> KErrorStreamX<K, V, K, VR> flatMapValuesCapturingErrorsInternal(
             final ValueMapperWithKey<K, V, Iterable<ProcessedValue<V, VR>>> mapper, final Named named) {
         final KStreamX<K, ProcessedValue<V, VR>> map = this.flatMapValues(mapper, named);
         return new ValueKErrorStreamX<>(map);
     }
 
-    private <KOut, VOut> KeyValueKErrorStreamX<K, V, KOut, VOut> processCapturingErrorsInternal(
+    private <KOut, VOut> KErrorStreamX<K, V, KOut, VOut> processCapturingErrorsInternal(
             final ProcessorSupplier<K, V, KOut, ProcessedKeyValue<K, V, VOut>> processorSupplier,
             final String... stateStoreNames) {
         final KStreamX<KOut, ProcessedKeyValue<K, V, VOut>> map =
@@ -1017,7 +1019,7 @@ private <KOut, VOut> KeyValueKErrorStreamX<K, V, KOut, VOut> processCapturingErr
         return new KeyValueKErrorStreamX<>(map);
     }
 
-    private <KOut, VOut> KeyValueKErrorStreamX<K, V, KOut, VOut> processCapturingErrorsInternal(
+    private <KOut, VOut> KErrorStreamX<K, V, KOut, VOut> processCapturingErrorsInternal(
             final ProcessorSupplier<K, V, KOut, ProcessedKeyValue<K, V, VOut>> processorSupplier, final Named named,
             final String... stateStoreNames) {
         final KStreamX<KOut, ProcessedKeyValue<K, V, VOut>> map =
@@ -1025,14 +1027,14 @@ private <KOut, VOut> KeyValueKErrorStreamX<K, V, KOut, VOut> processCapturingErr
         return new KeyValueKErrorStreamX<>(map);
     }
 
-    private <VOut> ValueKErrorStreamX<K, V, VOut> processValuesCapturingErrorsInternal(
+    private <VOut> KErrorStreamX<K, V, K, VOut> processValuesCapturingErrorsInternal(
             final FixedKeyProcessorSupplier<? super K, V, ProcessedValue<V, VOut>> processorSupplier,
             final String... stateStoreNames) {
         final KStreamX<K, ProcessedValue<V, VOut>> map = this.processValues(processorSupplier, stateStoreNames);
         return new ValueKErrorStreamX<>(map);
     }
 
-    private <VOut> ValueKErrorStreamX<K, V, VOut> processValuesCapturingErrorsInternal(
+    private <VOut> KErrorStreamX<K, V, K, VOut> processValuesCapturingErrorsInternal(
             final FixedKeyProcessorSupplier<? super K, V, ProcessedValue<V, VOut>> processorSupplier,
             final Named named, final String... stateStoreNames) {
         final KStreamX<K, ProcessedValue<V, VOut>> map =
diff --git a/streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KTableX.java b/streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KTableX.java
index 3a77d5c8..833f3334 100644
--- a/streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KTableX.java
+++ b/streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KTableX.java
@@ -205,16 +205,19 @@ <VR> KTableX<K, VR> transformValues(
             String... stateStoreNames);
 
     @Override
-    <KR, VR> KGroupedTableX<KR, VR> groupBy(KeyValueMapper<? super K, ? super V, KeyValue<KR, VR>> selector);
+    <KR, VR> KGroupedTableX<KR, VR> groupBy(
+            KeyValueMapper<? super K, ? super V, ? extends KeyValue<? extends KR, ? extends VR>> selector);
 
     @Override
-    <KR, VR> KGroupedTableX<KR, VR> groupBy(KeyValueMapper<? super K, ? super V, KeyValue<KR, VR>> selector,
+    <KR, VR> KGroupedTableX<KR, VR> groupBy(
+            KeyValueMapper<? super K, ? super V, ? extends KeyValue<? extends KR, ? extends VR>> selector,
             Grouped<KR, VR> grouped);
 
     /**
      * @see #groupBy(KeyValueMapper, Grouped)
      */
-    <KR, VR> KGroupedTableX<KR, VR> groupBy(KeyValueMapper<? super K, ? super V, KeyValue<KR, VR>> selector,
+    <KR, VR> KGroupedTableX<KR, VR> groupBy(
+            KeyValueMapper<? super K, ? super V, ? extends KeyValue<? extends KR, ? extends VR>> selector,
             GroupedX<KR, VR> grouped);
 
     @Override
@@ -311,122 +314,142 @@ <VO, VR> KTableX<K, VR> outerJoin(KTable<K, VO> other,
             Named named, MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 
     @Override
-    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, Function<V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner);
+    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, Function<? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner);
 
     @Override
-    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, Function<V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, TableJoined<K, KO> tableJoined);
+    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, Function<? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner, TableJoined<K, KO> tableJoined);
 
     @Override
-    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, Function<V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
+    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, Function<? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
+            Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 
     /**
      * @see #join(KTable, Function, ValueJoiner, Materialized)
      */
-    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, Function<V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
+    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, Function<? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
+            MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 
     @Override
-    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, Function<V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, TableJoined<K, KO> tableJoined,
+    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, Function<? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner, TableJoined<K, KO> tableJoined,
             Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 
     /**
      * @see #join(KTable, Function, ValueJoiner, TableJoined, Materialized)
      */
-    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, Function<V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, TableJoined<K, KO> tableJoined,
+    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, Function<? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner, TableJoined<K, KO> tableJoined,
             MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 
     @Override
-    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, Function<V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner);
+    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, Function<? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner);
 
     @Override
-    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, Function<V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, TableJoined<K, KO> tableJoined);
+    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, Function<? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner, TableJoined<K, KO> tableJoined);
 
     @Override
-    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, Function<V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
+    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, Function<? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
+            Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 
     /**
      * @see #leftJoin(KTable, Function, ValueJoiner, Materialized)
      */
-    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, Function<V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
+    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, Function<? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
+            MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 
     @Override
-    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, Function<V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, TableJoined<K, KO> tableJoined,
+    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, Function<? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner, TableJoined<K, KO> tableJoined,
             Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 
     /**
      * @see #leftJoin(KTable, Function, ValueJoiner, TableJoined, Materialized)
      */
-    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, Function<V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, TableJoined<K, KO> tableJoined,
+    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, Function<? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner, TableJoined<K, KO> tableJoined,
             MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 
     @Override
-    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, BiFunction<K, V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner);
+    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other,
+            BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner);
 
     @Override
-    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, BiFunction<K, V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, TableJoined<K, KO> tableJoined);
+    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other,
+            BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner, TableJoined<K, KO> tableJoined);
 
     @Override
-    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, BiFunction<K, V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
+    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other,
+            BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
+            Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 
     /**
      * @see #join(KTable, BiFunction, ValueJoiner, Materialized)
      */
-    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, BiFunction<K, V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
+    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other,
+            BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
+            MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 
     @Override
-    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, BiFunction<K, V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, TableJoined<K, KO> tableJoined,
+    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other,
+            BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner, TableJoined<K, KO> tableJoined,
             Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 
     /**
      * @see #join(KTable, BiFunction, ValueJoiner, TableJoined, Materialized)
      */
-    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other, BiFunction<K, V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, TableJoined<K, KO> tableJoined,
+    <VR, KO, VO> KTableX<K, VR> join(KTable<KO, VO> other,
+            BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner, TableJoined<K, KO> tableJoined,
             MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 
     @Override
-    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, BiFunction<K, V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner);
+    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other,
+            BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner);
 
     @Override
-    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, BiFunction<K, V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, TableJoined<K, KO> tableJoined);
+    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other,
+            BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner, TableJoined<K, KO> tableJoined);
 
     @Override
-    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, BiFunction<K, V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
+    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other,
+            BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
+            Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 
     /**
      * @see #leftJoin(KTable, BiFunction, ValueJoiner, Materialized)
      */
-    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, BiFunction<K, V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
+    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other,
+            BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
+            MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 
     @Override
-    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, BiFunction<K, V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, TableJoined<K, KO> tableJoined,
+    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other,
+            BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner, TableJoined<K, KO> tableJoined,
             Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 
     /**
      * @see #leftJoin(KTable, BiFunction, ValueJoiner, TableJoined, Materialized)
      */
-    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other, BiFunction<K, V, KO> foreignKeyExtractor,
-            ValueJoiner<V, VO, VR> joiner, TableJoined<K, KO> tableJoined,
+    <VR, KO, VO> KTableX<K, VR> leftJoin(KTable<KO, VO> other,
+            BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            ValueJoiner<? super V, ? super VO, ? extends VR> joiner, TableJoined<K, KO> tableJoined,
             MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 }
diff --git a/streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KTableXImpl.java b/streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KTableXImpl.java
index 34b7c250..2af2e71b 100644
--- a/streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KTableXImpl.java
+++ b/streams-bootstrap-core/src/main/java/com/bakdata/kafka/streams/kstream/KTableXImpl.java
@@ -267,19 +267,21 @@ public <VR> KTableX<K, VR> transformValues(
 
     @Override
     public <KR, VR> KGroupedTableX<KR, VR> groupBy(
-            final KeyValueMapper<? super K, ? super V, KeyValue<KR, VR>> selector) {
+            final KeyValueMapper<? super K, ? super V, ? extends KeyValue<? extends KR, ? extends VR>> selector) {
         return this.context.wrap(this.wrapped.groupBy(selector));
     }
 
     @Override
     public <KR, VR> KGroupedTableX<KR, VR> groupBy(
-            final KeyValueMapper<? super K, ? super V, KeyValue<KR, VR>> selector, final Grouped<KR, VR> grouped) {
+            final KeyValueMapper<? super K, ? super V, ? extends KeyValue<? extends KR, ? extends VR>> selector,
+            final Grouped<KR, VR> grouped) {
         return this.context.wrap(this.wrapped.groupBy(selector, grouped));
     }
 
     @Override
     public <KR, VR> KGroupedTableX<KR, VR> groupBy(
-            final KeyValueMapper<? super K, ? super V, KeyValue<KR, VR>> selector, final GroupedX<KR, VR> grouped) {
+            final KeyValueMapper<? super K, ? super V, ? extends KeyValue<? extends KR, ? extends VR>> selector,
+            final GroupedX<KR, VR> grouped) {
         return this.groupBy(selector, grouped.configure(this.context.getConfigurator()));
     }
 
@@ -417,40 +419,41 @@ public <VO, VR> KTableX<K, VR> outerJoin(final KTable<K, VO> other,
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other,
-            final Function<V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner) {
+            final Function<? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner) {
         final KTable<KO, VO> otherTable = StreamsContext.maybeUnwrap(other);
         return this.context.wrap(this.wrapped.join(otherTable, foreignKeyExtractor, joiner));
     }
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other,
-            final Function<V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner, final TableJoined<K, KO> tableJoined) {
+            final Function<? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner, final TableJoined<K, KO> tableJoined) {
         final KTable<KO, VO> otherTable = StreamsContext.maybeUnwrap(other);
         return this.context.wrap(this.wrapped.join(otherTable, foreignKeyExtractor, joiner, tableJoined));
     }
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other,
-            final Function<V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner, final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
+            final Function<? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
+            final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
         final KTable<KO, VO> otherTable = StreamsContext.maybeUnwrap(other);
         return this.context.wrap(this.wrapped.join(otherTable, foreignKeyExtractor, joiner, materialized));
     }
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other,
-            final Function<V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner,
+            final Function<? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
             final MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
         return this.join(other, foreignKeyExtractor, joiner, materialized.configure(this.context.getConfigurator()));
     }
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other,
-            final Function<V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner, final TableJoined<K, KO> tableJoined,
+            final Function<? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner, final TableJoined<K, KO> tableJoined,
             final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
         final KTable<KO, VO> otherTable = StreamsContext.maybeUnwrap(other);
         return this.context.wrap(this.wrapped.join(otherTable, foreignKeyExtractor, joiner, tableJoined, materialized));
@@ -458,8 +461,8 @@ public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other,
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other,
-            final Function<V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner, final TableJoined<K, KO> tableJoined,
+            final Function<? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner, final TableJoined<K, KO> tableJoined,
             final MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
         return this.join(other, foreignKeyExtractor, joiner, tableJoined,
                 materialized.configure(this.context.getConfigurator()));
@@ -467,32 +470,33 @@ public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other,
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> leftJoin(final KTable<KO, VO> other,
-            final Function<V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner) {
+            final Function<? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner) {
         final KTable<KO, VO> otherTable = StreamsContext.maybeUnwrap(other);
         return this.context.wrap(this.wrapped.leftJoin(otherTable, foreignKeyExtractor, joiner));
     }
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> leftJoin(final KTable<KO, VO> other,
-            final Function<V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner, final TableJoined<K, KO> tableJoined) {
+            final Function<? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner, final TableJoined<K, KO> tableJoined) {
         final KTable<KO, VO> otherTable = StreamsContext.maybeUnwrap(other);
         return this.context.wrap(this.wrapped.leftJoin(otherTable, foreignKeyExtractor, joiner, tableJoined));
     }
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> leftJoin(final KTable<KO, VO> other,
-            final Function<V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner, final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
+            final Function<? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
+            final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
         final KTable<KO, VO> otherTable = StreamsContext.maybeUnwrap(other);
         return this.context.wrap(this.wrapped.leftJoin(otherTable, foreignKeyExtractor, joiner, materialized));
     }
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> leftJoin(final KTable<KO, VO> other,
-            final Function<V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner,
+            final Function<? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
             final MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
         return this.leftJoin(other, foreignKeyExtractor, joiner,
                 materialized.configure(this.context.getConfigurator()));
@@ -500,8 +504,8 @@ public <VR, KO, VO> KTableX<K, VR> leftJoin(final KTable<KO, VO> other,
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> leftJoin(final KTable<KO, VO> other,
-            final Function<V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner, final TableJoined<K, KO> tableJoined,
+            final Function<? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner, final TableJoined<K, KO> tableJoined,
             final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
         final KTable<KO, VO> otherTable = StreamsContext.maybeUnwrap(other);
         return this.context.wrap(
@@ -510,52 +514,59 @@ public <VR, KO, VO> KTableX<K, VR> leftJoin(final KTable<KO, VO> other,
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> leftJoin(final KTable<KO, VO> other,
-            final Function<V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner, final TableJoined<K, KO> tableJoined,
+            final Function<? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner, final TableJoined<K, KO> tableJoined,
             final MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
         return this.leftJoin(other, foreignKeyExtractor, joiner, tableJoined,
                 materialized.configure(this.context.getConfigurator()));
     }
 
     @Override
-    public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other, final BiFunction<K, V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner) {
+    public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other,
+            final BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner) {
         final KTable<KO, VO> otherTable = StreamsContext.maybeUnwrap(other);
         return this.context.wrap(this.wrapped.join(otherTable, foreignKeyExtractor, joiner));
     }
 
     @Override
-    public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other, final BiFunction<K, V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner, final TableJoined<K, KO> tableJoined) {
+    public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other,
+            final BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner, final TableJoined<K, KO> tableJoined) {
         final KTable<KO, VO> otherTable = StreamsContext.maybeUnwrap(other);
         return this.context.wrap(this.wrapped.join(otherTable, foreignKeyExtractor, joiner, tableJoined));
     }
 
     @Override
-    public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other, final BiFunction<K, V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner, final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
+    public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other,
+            final BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
+            final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
         final KTable<KO, VO> otherTable = StreamsContext.maybeUnwrap(other);
         return this.context.wrap(this.wrapped.join(otherTable, foreignKeyExtractor, joiner, materialized));
     }
 
     @Override
-    public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other, final BiFunction<K, V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner,
+    public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other,
+            final BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
             final MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
         return this.join(other, foreignKeyExtractor, joiner, materialized.configure(this.context.getConfigurator()));
     }
 
     @Override
-    public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other, final BiFunction<K, V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner, final TableJoined<K, KO> tableJoined,
+    public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other,
+            final BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner, final TableJoined<K, KO> tableJoined,
             final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
         final KTable<KO, VO> otherTable = StreamsContext.maybeUnwrap(other);
         return this.context.wrap(this.wrapped.join(otherTable, foreignKeyExtractor, joiner, tableJoined, materialized));
     }
 
     @Override
-    public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other, final BiFunction<K, V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner, final TableJoined<K, KO> tableJoined,
+    public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other,
+            final BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner, final TableJoined<K, KO> tableJoined,
             final MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
         return this.join(other, foreignKeyExtractor, joiner, tableJoined,
                 materialized.configure(this.context.getConfigurator()));
@@ -563,32 +574,33 @@ public <VR, KO, VO> KTableX<K, VR> join(final KTable<KO, VO> other, final BiFunc
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> leftJoin(final KTable<KO, VO> other,
-            final BiFunction<K, V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner) {
+            final BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner) {
         final KTable<KO, VO> otherTable = StreamsContext.maybeUnwrap(other);
         return this.context.wrap(this.wrapped.leftJoin(otherTable, foreignKeyExtractor, joiner));
     }
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> leftJoin(final KTable<KO, VO> other,
-            final BiFunction<K, V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner, final TableJoined<K, KO> tableJoined) {
+            final BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner, final TableJoined<K, KO> tableJoined) {
         final KTable<KO, VO> otherTable = StreamsContext.maybeUnwrap(other);
         return this.context.wrap(this.wrapped.leftJoin(otherTable, foreignKeyExtractor, joiner, tableJoined));
     }
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> leftJoin(final KTable<KO, VO> other,
-            final BiFunction<K, V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner, final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
+            final BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
+            final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
         final KTable<KO, VO> otherTable = StreamsContext.maybeUnwrap(other);
         return this.context.wrap(this.wrapped.leftJoin(otherTable, foreignKeyExtractor, joiner, materialized));
     }
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> leftJoin(final KTable<KO, VO> other,
-            final BiFunction<K, V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner,
+            final BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner,
             final MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
         return this.leftJoin(other, foreignKeyExtractor, joiner,
                 materialized.configure(this.context.getConfigurator()));
@@ -596,8 +608,8 @@ public <VR, KO, VO> KTableX<K, VR> leftJoin(final KTable<KO, VO> other,
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> leftJoin(final KTable<KO, VO> other,
-            final BiFunction<K, V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner, final TableJoined<K, KO> tableJoined,
+            final BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner, final TableJoined<K, KO> tableJoined,
             final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
         final KTable<KO, VO> otherTable = StreamsContext.maybeUnwrap(other);
         return this.context.wrap(
@@ -606,8 +618,8 @@ public <VR, KO, VO> KTableX<K, VR> leftJoin(final KTable<KO, VO> other,
 
     @Override
     public <VR, KO, VO> KTableX<K, VR> leftJoin(final KTable<KO, VO> other,
-            final BiFunction<K, V, KO> foreignKeyExtractor,
-            final ValueJoiner<V, VO, VR> joiner, final TableJoined<K, KO> tableJoined,
+            final BiFunction<? super K, ? super V, ? extends KO> foreignKeyExtractor,
+            final ValueJoiner<? super V, ? super VO, ? extends VR> joiner, final TableJoined<K, KO> tableJoined,
             final MaterializedX<K, VR, KeyValueStore<Bytes, byte[]>> materialized) {
         return this.leftJoin(other, foreignKeyExtractor, joiner, tableJoined,
                 materialized.configure(this.context.getConfigurator()));