diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java
index b23d51393..11bd7874d 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java
@@ -16,15 +16,9 @@
 
 package org.springframework.cloud.stream.binder.kafka.streams;
 
-import java.lang.reflect.Method;
-import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.Optional;
-import java.util.UUID;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.common.utils.Utils;
@@ -33,7 +27,6 @@ import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.KTable;
 
 import org.springframework.beans.BeansException;
-import org.springframework.beans.factory.annotation.AnnotatedBeanDefinition;
 import org.springframework.cloud.stream.binder.ConsumerProperties;
 import org.springframework.cloud.stream.binder.ProducerProperties;
 import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties;
@@ -43,8 +36,6 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.core.ResolvableType;
-import org.springframework.kafka.support.serializer.JsonSerde;
-import org.springframework.util.ClassUtils;
 import org.springframework.util.StringUtils;
 
 /**
@@ -71,11 +62,10 @@ import org.springframework.util.StringUtils;
  * @author Soby Chacko
  * @author Lei Chen
  * @author Eduard Domínguez
+ * @author Chris Bono
  */
 public class KeyValueSerdeResolver implements ApplicationContextAware {
 
-	private static final Log LOG = LogFactory.getLog(KeyValueSerdeResolver.class);
-
 	private final Map<String, Object> streamConfigGlobalProperties;
 
 	private final KafkaStreamsBinderConfigurationProperties binderConfigurationProperties;
@@ -262,7 +252,7 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
 						(isResolvalbeKafkaStreamsType(resolvableType) || isResolvableKStreamArrayType(resolvableType))) {
 					ResolvableType generic = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(0) : resolvableType.getGeneric(0);
 					Serde<?> fallbackSerde = getFallbackSerde("default.key.serde");
-					keySerde = getSerde(generic, fallbackSerde);
+					keySerde = SerdeResolverUtils.resolveForType(this.context, generic, fallbackSerde);
 				}
 				if (keySerde == null) {
 					keySerde = Serdes.ByteArray();
@@ -286,101 +276,6 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
 				GlobalKTable.class.isAssignableFrom(resolvableType.getRawClass()));
 	}
 
-	private Serde<?> getSerde(ResolvableType generic, Serde<?> fallbackSerde) {
-		Serde<?> serde = null;
-
-		Map<String, Serde> beansOfType = context.getBeansOfType(Serde.class);
-		Serde<?>[] serdeBeans = new Serde<?>[1];
-
-		final Class<?> genericRawClazz = generic.getRawClass();
-		beansOfType.forEach((k, v) -> {
-			final Class<?> classObj = ClassUtils.resolveClassName(((AnnotatedBeanDefinition)
-							context.getBeanFactory().getBeanDefinition(k))
-							.getMetadata().getClassName(),
-					ClassUtils.getDefaultClassLoader());
-			try {
-				Method[] methods = classObj.getMethods();
-				Optional<Method> serdeBeanMethod = Arrays.stream(methods).filter(m -> m.getName().equals(k)).findFirst();
-				if (serdeBeanMethod.isPresent()) {
-					Method method = serdeBeanMethod.get();
-					ResolvableType resolvableType = ResolvableType.forMethodReturnType(method, classObj);
-					ResolvableType serdeBeanGeneric = resolvableType.getGeneric(0);
-					Class<?> serdeGenericRawClazz = serdeBeanGeneric.getRawClass();
-					if (serdeGenericRawClazz != null && genericRawClazz != null) {
-						if (serdeGenericRawClazz.isAssignableFrom(genericRawClazz)) {
-							serdeBeans[0] = v;
-						}
-					}
-				}
-			}
-			catch (Exception e) {
-				// Pass through...
-			}
-
-		});
-
-		if (serdeBeans[0] != null) {
-			return serdeBeans[0];
-		}
-
-		if (genericRawClazz != null) {
-			if (Integer.class.isAssignableFrom(genericRawClazz)) {
-				serde = Serdes.Integer();
-			}
-			else if (Long.class.isAssignableFrom(genericRawClazz)) {
-				serde = Serdes.Long();
-			}
-			else if (Short.class.isAssignableFrom(genericRawClazz)) {
-				serde = Serdes.Short();
-			}
-			else if (Double.class.isAssignableFrom(genericRawClazz)) {
-				serde = Serdes.Double();
-			}
-			else if (Float.class.isAssignableFrom(genericRawClazz)) {
-				serde = Serdes.Float();
-			}
-			else if (byte[].class.isAssignableFrom(genericRawClazz)) {
-				serde = Serdes.ByteArray();
-			}
-			else if (String.class.isAssignableFrom(genericRawClazz)) {
-				serde = Serdes.String();
-			}
-			else if (UUID.class.isAssignableFrom(genericRawClazz)) {
-				serde = Serdes.UUID();
-			}
-			else if (!isSerdeFromStandardDefaults(fallbackSerde)) {
-				//User purposely set a default serde that is not one of the above
-				serde = fallbackSerde;
-			}
-			else {
-				// If the type is Object, then skip assigning the JsonSerde and let the fallback mechanism takes precedence.
-				if (!genericRawClazz.isAssignableFrom((Object.class))) {
-					serde = new JsonSerde(genericRawClazz);
-				}
-			}
-		}
-		return serde;
-	}
-
-	private boolean isSerdeFromStandardDefaults(Serde<?> serde) {
-		if (serde != null) {
-			if (Number.class.isAssignableFrom(serde.getClass())) {
-				return true;
-			}
-			else if (Serdes.ByteArray().getClass().isAssignableFrom(serde.getClass())) {
-				return true;
-			}
-			else if (Serdes.String().getClass().isAssignableFrom(serde.getClass())) {
-				return true;
-			}
-			else if (Serdes.UUID().getClass().isAssignableFrom(serde.getClass())) {
-				return true;
-			}
-		}
-		return false;
-	}
-
-
 	private Serde<?> getValueSerde(String valueSerdeString, Map<String, ?> extendedConfiguration)
 			throws ClassNotFoundException {
 		Serde<?> valueSerde;
@@ -416,7 +311,7 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
 					(isResolvableKStreamArrayType(resolvableType)))) {
 				Serde<?> fallbackSerde = getFallbackSerde("default.value.serde");
 				ResolvableType generic = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(1) : resolvableType.getGeneric(1);
-				valueSerde = getSerde(generic, fallbackSerde);
+				valueSerde = SerdeResolverUtils.resolveForType(this.context, generic, fallbackSerde);
 			}
 			if (valueSerde == null) {
 
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtils.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtils.java
new file mode 100644
index 000000000..3509bc29b
--- /dev/null
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtils.java
@@ -0,0 +1,241 @@
+/*
+ * Copyright 2022-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka.streams;
+
+import java.lang.reflect.Method;
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serdes;
+
+import org.springframework.beans.factory.annotation.AnnotatedBeanDefinition;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.core.ResolvableType;
+import org.springframework.kafka.support.serializer.JsonSerde;
+import org.springframework.util.ClassUtils;
+
+/**
+ * Utility class that contains various methods to help resolve {@link Serde Serdes}.
+ *
+ * @author Chris Bono
+ * @since 4.0
+ */
+abstract class SerdeResolverUtils {
+
+	private static final Log LOG = LogFactory.getLog(SerdeResolverUtils.class);
+
+	/**
+	 *  Return the closest matching configured {@code Serde<?>} bean if one exists, or the specified
+	 *  {@code fallbackSerde}, or finally a standard default serde if no fallback specified.
+	 *
+	 * @param context the application context
+	 * @param targetType the target type to find the serde for
+	 * @param fallbackSerde the fallback serde in case no matching serde bean found in the context
+	 * @return serde to use for the target type
+	 */
+	static Serde<?> resolveForType(ConfigurableApplicationContext context, ResolvableType targetType, Serde<?> fallbackSerde) {
+
+		List<Serde<?>> matchingSerdes = findMatchingSerdes(context, targetType);
+		if (!matchingSerdes.isEmpty()) {
+			return matchingSerdes.get(0);
+		}
+
+		// We don't attempt to find a matching Serde for type '?'
+		if (targetType.getRawClass() == null) {
+			return null;
+		}
+
+		Serde<?> serde = null;
+		Class<?> genericRawClazz = targetType.getRawClass();
+		if (Integer.class.isAssignableFrom(genericRawClazz)) {
+			serde = Serdes.Integer();
+		}
+		else if (Long.class.isAssignableFrom(genericRawClazz)) {
+			serde = Serdes.Long();
+		}
+		else if (Short.class.isAssignableFrom(genericRawClazz)) {
+			serde = Serdes.Short();
+		}
+		else if (Double.class.isAssignableFrom(genericRawClazz)) {
+			serde = Serdes.Double();
+		}
+		else if (Float.class.isAssignableFrom(genericRawClazz)) {
+			serde = Serdes.Float();
+		}
+		else if (byte[].class.isAssignableFrom(genericRawClazz)) {
+			serde = Serdes.ByteArray();
+		}
+		else if (String.class.isAssignableFrom(genericRawClazz)) {
+			serde = Serdes.String();
+		}
+		else if (UUID.class.isAssignableFrom(genericRawClazz)) {
+			serde = Serdes.UUID();
+		}
+		else if (!isSerdeFromStandardDefaults(fallbackSerde)) {
+			// User purposely set a default serde that is not one of the above
+			serde = fallbackSerde;
+		}
+		else {
+			// If the type is Object, then skip assigning the JsonSerde and let the fallback mechanism takes precedence.
+			if (!genericRawClazz.isAssignableFrom((Object.class))) {
+				serde = new JsonSerde(genericRawClazz);
+			}
+		}
+		return serde;
+	}
+
+	private static boolean isSerdeFromStandardDefaults(Serde<?> serde) {
+		if (serde != null) {
+			if (Number.class.isAssignableFrom(serde.getClass())) {
+				return true;
+			}
+			else if (Serdes.ByteArray().getClass().isAssignableFrom(serde.getClass())) {
+				return true;
+			}
+			else if (Serdes.String().getClass().isAssignableFrom(serde.getClass())) {
+				return true;
+			}
+			else if (Serdes.UUID().getClass().isAssignableFrom(serde.getClass())) {
+				return true;
+			}
+		}
+		return false;
+	}
+
+	/**
+	 * Find all {@link Serde} beans that are assignable from {@code targetType}.
+	 *
+	 * @param context the application context
+	 * @param targetType the target type the serdes are being matched for
+	 * @return list of matching serdes order by most specific match, or an empty list if no matches found
+	 */
+	static List<Serde<?>> findMatchingSerdes(ConfigurableApplicationContext context, ResolvableType targetType) {
+		// We don't attempt to find a matching Serde for type '?'
+		if (targetType.getRawClass() == null) {
+			return Collections.emptyList();
+		}
+		List<SerdeWithSpecificityScore> matchingSerdes = new ArrayList<>();
+
+		context.getBeansOfType(Serde.class).forEach((beanName, serdeBean) -> {
+			final Class<?> beanConfigClass = ClassUtils.resolveClassName(((AnnotatedBeanDefinition)
+					context.getBeanFactory().getBeanDefinition(beanName))
+					.getMetadata().getClassName(),
+				ClassUtils.getDefaultClassLoader());
+			try {
+				Method[] methods = beanConfigClass.getMethods();
+				Optional<Method> serdeBeanMethod = Arrays.stream(methods).filter(m -> m.getName().equals(beanName)).findFirst();
+				serdeBeanMethod.ifPresent((method) -> {
+					ResolvableType serdeBeanMethodReturnType = ResolvableType.forMethodReturnType(method, beanConfigClass);
+					ResolvableType serdeBeanGeneric = serdeBeanMethodReturnType.getGeneric(0);
+					// We don't attempt to use a Serde<?> as a match for anything currently
+					if (serdeBeanGeneric.getRawClass() != null && serdeBeanGeneric.isAssignableFrom(targetType)) {
+						matchingSerdes.add(new SerdeWithSpecificityScore(calculateScore(targetType, serdeBeanGeneric), serdeBean));
+					}
+				});
+			}
+			catch (Exception e) {
+				if (LOG.isTraceEnabled()) {
+					LOG.trace("Failed to introspect Serde bean method '" + serdeBean + "'", e);
+				}
+			}
+		});
+		if (!matchingSerdes.isEmpty()) {
+			return matchingSerdes.stream().sorted(Collections.reverseOrder())
+				.map(SerdeWithSpecificityScore::getSerde)
+				.collect(Collectors.toList());
+		}
+		return Collections.emptyList();
+	}
+
+	/**
+	 * Calculate a score to indicate how specific of a match one resolvable type is to another.
+	 * <p><br>Simple string comparison (the number of matching leading characters between two type strings) is used to
+	 * calculate the score. This approach avoids the recursive nature of the possible generic types, and leverages
+	 * the type strings returned from {@link ResolvableType#toString()} and {@link Type#getTypeName()} which already
+	 * include the properly handled generic types. The score is a composite of both of these properties because the
+	 * 'toString' value does not include bounds values and works as a tie-breaker to distinguish non-exact
+	 * matches that are closer in nature.
+	 * <p><br><b>Example:</b>
+	 * <pre>{@code
+	 * -------------------------------------------------------------------------------------------------------
+	 * targetType: Foo<Date>               toString='Foo<Date>'     typeName='Foo<java.util.Date>'
+	 * typeToCheck1: Foo<Date>             toString='Foo<Date>'     typeName='Foo<java.util.Date>'
+	 * typeToCheck2: Foo<? extends Date>   toString='Foo<Date>'     typeName='Foo<? extends java.util.Date>'
+	 * -------------------------------------------------------------------------------------------------------
+	 * }</pre>
+	 *
+	 * If using only the 'toString' value then both types would have the same score. However, including the 'typeName'
+	 * value in the score differentiates them - in this case it is clear that 'typeToCheck1' is a direct match.
+	 *
+	 * @param targetType the target type
+	 * @param typeToScore the type to calculate a score for
+	 * @return a score on how close of a match {@code typeToScore} is to {@code targetType} - the higher the score,
+	 * 	the closer the match
+	 */
+	private static int calculateScore(ResolvableType targetType, ResolvableType typeToScore) {
+		int score = countLeadingMatchingChars(targetType.getType().getTypeName(), typeToScore.getType().getTypeName());
+		score += countLeadingMatchingChars(targetType.toString(), typeToScore.toString());
+		return score;
+	}
+
+	private static int countLeadingMatchingChars(String s1, String s2) {
+		if (s1 == null || s2 == null) {
+			return 0;
+		}
+		int matchCount = 0;
+		for (int i = 0; i < s1.length() && i < s2.length(); i++) {
+			if (s1.charAt(i) != s2.charAt(i)) {
+				break;
+			}
+			matchCount++;
+		}
+		return matchCount;
+	}
+
+	/**
+	 * Private internal class used strictly to 'remember' a score for a serde and use it for sorting later.
+	 */
+	private static class SerdeWithSpecificityScore implements Comparable<SerdeWithSpecificityScore> {
+		private Integer score;
+		private Serde<?> serde;
+
+		SerdeWithSpecificityScore(Integer score, Serde<?> serde) {
+			this.score = Objects.requireNonNull(score);
+			this.serde = Objects.requireNonNull(serde);
+		}
+
+		Serde<?> getSerde() {
+			return serde;
+		}
+
+		@Override
+		public int compareTo(SerdeWithSpecificityScore other) {
+			return this.score.compareTo(other.score);
+		}
+	}
+
+}
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtilsTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtilsTests.java
new file mode 100644
index 000000000..f59e28b50
--- /dev/null
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtilsTests.java
@@ -0,0 +1,149 @@
+/*
+ * Copyright 2022-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka.streams;
+
+import java.util.Date;
+
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.Serializer;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.context.annotation.Bean;
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.core.ResolvableType;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Unit tests for {@link SerdeResolverUtils}.
+ *
+ * @author Chris Bono
+ */
+class SerdeResolverUtilsTests {
+
+	/**
+	 * Verify that {@link SerdeResolverUtils#findMatchingSerdes} returns the proper serdes
+	 * in the proper order for the following grid:
+	 * <p><br>
+	 * <pre>{@code
+	 * ------------------------------------------------------------------
+	 * KStream type       | Serde type
+	 * ------------------------------------------------------------------
+	 *                    | GE<Date> | GE<? extends Date> | GE<?> | GE
+	 * ------------------------------------------------------------------
+	 * GE<Date>           | Y        | Y                  | Y     | Y
+	 * GE<? extends Date> | N        | Y                  | Y     | N
+	 * GE<?>              | N        | N                  | Y     | N
+	 * GE                 | N        | N                  | N     | N
+	 * ------------------------------------------------------------------
+	 * }</pre>
+	 */
+	@Test
+	void findMatchingSerdesForSimpleGenericType() {
+
+		ResolvableType geDate = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<Date>>() { });
+		ResolvableType geBounded = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<? extends Date>>() { });
+		ResolvableType geWildcard = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<?>>() { });
+		ResolvableType geRaw = ResolvableType.forRawClass(GenericEvent.class);
+
+		ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+			.withUserConfiguration(SerdeResolverTestApp.class);
+
+		contextRunner.run((context) -> {
+
+			assertThat(SerdeResolverUtils.findMatchingSerdes(context, geDate))
+				.extracting("name")
+				.containsExactly("genericEventDateSerde", "genericEventDateBoundedSerde", "genericEventWildcardSerde", "genericEventRawSerde");
+
+			assertThat(SerdeResolverUtils.findMatchingSerdes(context, geBounded))
+				.extracting("name")
+				.containsExactly("genericEventDateBoundedSerde", "genericEventWildcardSerde");
+
+			assertThat(SerdeResolverUtils.findMatchingSerdes(context, geWildcard))
+				.extracting("name")
+				.containsExactly("genericEventWildcardSerde");
+
+			// Because GenericEvent is a parameterized type, Serde<GenericEvent> resolves to Serde<GenericEvent<?>>
+			// which is not assignable from GenericEvent
+			assertThat(SerdeResolverUtils.findMatchingSerdes(context, geRaw))
+				.extracting("name")
+				.isEmpty();
+		});
+	}
+
+	static class GenericEventSerde<T> implements Serde<GenericEvent<? extends T>> {
+		private String name;
+
+		GenericEventSerde(String name) {
+			this.name = name;
+		}
+
+		String getName() {
+			return name;
+		}
+
+		@Override
+		public Serializer<GenericEvent<? extends T>> serializer() {
+			return null;
+		}
+
+		@Override
+		public Deserializer<GenericEvent<? extends T>> deserializer() {
+			return null;
+		}
+	}
+
+	static class GenericEvent<T> { }
+
+	@EnableAutoConfiguration
+	static class SerdeResolverTestApp {
+
+		@Bean
+		public Serde<GenericEvent<Date>> genericEventDateSerde() {
+			return new GenericEventSerde("genericEventDateSerde");
+		}
+
+		@Bean
+		public Serde<GenericEvent<? extends Date>> genericEventDateBoundedSerde() {
+			return new GenericEventSerde("genericEventDateBoundedSerde");
+		}
+
+		@Bean
+		public Serde<GenericEvent<String>> genericEventStringSerde() {
+			return new GenericEventSerde("genericEventStringSerde");
+		}
+
+		@Bean
+		public Serde<GenericEvent<?>> genericEventWildcardSerde() {
+			return new GenericEventSerde("genericEventWildcardSerde");
+		}
+
+		@Bean
+		public Serde<GenericEvent> genericEventRawSerde() {
+			return new GenericEventSerde("genericEventRawSerde");
+		}
+
+		@Bean
+		public Serde<?> widlcardSerde() {
+			return Serdes.Void();
+		}
+	}
+}
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/SerdesProvidedAsBeansTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/SerdesProvidedAsBeansTests.java
index f5893ce84..d783e1ca7 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/SerdesProvidedAsBeansTests.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/SerdesProvidedAsBeansTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2019 the original author or authors.
+ * Copyright 2019-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@ import java.util.function.Function;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.kstream.KStream;
 import org.junit.ClassRule;
 import org.junit.Test;
@@ -42,8 +43,15 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.core.ResolvableType;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
-import org.springframework.util.Assert;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests to verify proper resolution of {@link Serde serdes} when they are provided as beans.
+ *
+ * @author Soby Chako
+ * @author Chris Bono
+ */
 public class SerdesProvidedAsBeansTests {
 
 	@ClassRule
@@ -52,44 +60,78 @@ public class SerdesProvidedAsBeansTests {
 	private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka();
 
 	@Test
-	public void testKstreamWordCountFunction() throws NoSuchMethodException {
-		SpringApplication app = new SpringApplication(SerdeProvidedAsBeanApp.class);
+	public void testSimpleSerdeBeansAreResolvedProperly() throws Exception {
+		SpringApplication app = new SpringApplication(SerdesProvidedAsBeansTestApp.class);
 		app.setWebApplicationType(WebApplicationType.NONE);
-
 		try (ConfigurableApplicationContext context = app.run(
-				"--server.port=0",
-				"--spring.jmx.enabled=false",
-				"--spring.cloud.stream.bindings.process-in-0.destination=purchases",
-				"--spring.cloud.stream.bindings.process-out-0.destination=coffee",
-				"--spring.cloud.stream.kafka.streams.binder.functions.process.applicationId=process-id-0",
-				"--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000",
-				"--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" +
-						"=org.apache.kafka.common.serialization.Serdes$StringSerde",
-				"--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" +
-						"=org.apache.kafka.common.serialization.Serdes$StringSerde",
-				"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
+			"--server.port=0",
+			"--spring.jmx.enabled=false",
+			"--spring.cloud.function.definition=simpleProcess",
+			"--spring.cloud.stream.bindings.simpleProcess-in-0.destination=topic1",
+			"--spring.cloud.stream.bindings.simpleProcess-out-0.destination=topic2",
+			"--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000",
+			"--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" +
+				"=org.apache.kafka.common.serialization.Serdes$StringSerde",
+			"--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" +
+				"=org.apache.kafka.common.serialization.Serdes$StringSerde",
+			"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
 
-			final Method method = SerdeProvidedAsBeanApp.class.getMethod("process");
+			Method method = SerdesProvidedAsBeansTestApp.class.getMethod("simpleProcess");
+			ResolvableType resolvableType = ResolvableType.forMethodReturnType(method, SerdesProvidedAsBeansTestApp.class);
 
-			ResolvableType resolvableType = ResolvableType.forMethodReturnType(method, SerdeProvidedAsBeanApp.class);
+			KeyValueSerdeResolver keyValueSerdeResolver = context.getBean(KeyValueSerdeResolver.class);
 
-			final KeyValueSerdeResolver keyValueSerdeResolver = context.getBean(KeyValueSerdeResolver.class);
-			final BindingServiceProperties bindingServiceProperties = context.getBean(BindingServiceProperties.class);
-			final KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties = context.getBean(KafkaStreamsExtendedBindingProperties.class);
+			BindingServiceProperties bindingServiceProperties = context.getBean(BindingServiceProperties.class);
+			ConsumerProperties consumerProperties = bindingServiceProperties.getBindingProperties("simpleProcess-in-0").getConsumer();
+			KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties = context.getBean(KafkaStreamsExtendedBindingProperties.class);
+			KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties = kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties("input");
+			Serde<?> inboundValueSerde = keyValueSerdeResolver.getInboundValueSerde(consumerProperties, kafkaStreamsConsumerProperties, resolvableType.getGeneric(0));
 
-			final ConsumerProperties consumerProperties = bindingServiceProperties.getBindingProperties("process-in-0").getConsumer();
-			final KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties = kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties("input");
-			kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties("input");
-			final Serde<?> inboundValueSerde = keyValueSerdeResolver.getInboundValueSerde(consumerProperties, kafkaStreamsConsumerProperties, resolvableType.getGeneric(0));
+			assertThat(inboundValueSerde).isInstanceOf(FooSerde.class);
 
-			Assert.isTrue(inboundValueSerde instanceof FooSerde, "Inbound Value Serde is not matched");
+			ProducerProperties producerProperties = bindingServiceProperties.getBindingProperties("simpleProcess-out-0").getProducer();
+			KafkaStreamsProducerProperties kafkaStreamsProducerProperties = kafkaStreamsExtendedBindingProperties.getExtendedProducerProperties("output");
+			Serde<?> outboundValueSerde = keyValueSerdeResolver.getOutboundValueSerde(producerProperties, kafkaStreamsProducerProperties, resolvableType.getGeneric(1));
 
-			final ProducerProperties producerProperties = bindingServiceProperties.getBindingProperties("process-out-0").getProducer();
-			final KafkaStreamsProducerProperties kafkaStreamsProducerProperties = kafkaStreamsExtendedBindingProperties.getExtendedProducerProperties("output");
-			kafkaStreamsExtendedBindingProperties.getExtendedProducerProperties("output");
-			final Serde<?> outboundValueSerde = keyValueSerdeResolver.getOutboundValueSerde(producerProperties, kafkaStreamsProducerProperties, resolvableType.getGeneric(1));
+			assertThat(outboundValueSerde).isInstanceOf(FooSerde.class);
+		}
+	}
 
-			Assert.isTrue(outboundValueSerde instanceof FooSerde, "Outbound Value Serde is not matched");
+	@Test
+	public void testGenericSerdeBeansAreResolvedProperly() throws Exception {
+		SpringApplication app = new SpringApplication(SerdesProvidedAsBeansTestApp.class);
+		app.setWebApplicationType(WebApplicationType.NONE);
+		try (ConfigurableApplicationContext context = app.run(
+			"--server.port=0",
+			"--spring.jmx.enabled=false",
+			"--spring.cloud.function.definition=genericProcess",
+			"--spring.cloud.stream.bindings.genericProcess-in-0.destination=topic1",
+			"--spring.cloud.stream.bindings.genericProcess-out-0.destination=topic2",
+			"--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000",
+			"--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" +
+				"=org.apache.kafka.common.serialization.Serdes$StringSerde",
+			"--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" +
+				"=org.apache.kafka.common.serialization.Serdes$StringSerde",
+			"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
+
+			Method method = SerdesProvidedAsBeansTestApp.class.getMethod("genericProcess");
+			ResolvableType resolvableType = ResolvableType.forMethodReturnType(method, SerdesProvidedAsBeansTestApp.class);
+
+			KeyValueSerdeResolver keyValueSerdeResolver = context.getBean(KeyValueSerdeResolver.class);
+
+			BindingServiceProperties bindingServiceProperties = context.getBean(BindingServiceProperties.class);
+			ConsumerProperties consumerProperties = bindingServiceProperties.getBindingProperties("genericProcess-in-0").getConsumer();
+			KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties = context.getBean(KafkaStreamsExtendedBindingProperties.class);
+			KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties = kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties("input");
+			Serde<?> inboundValueSerde = keyValueSerdeResolver.getInboundValueSerde(consumerProperties, kafkaStreamsConsumerProperties, resolvableType.getGeneric(0));
+
+			assertThat(inboundValueSerde).isInstanceOf(GenericEventDateSerde.class);
+
+			ProducerProperties producerProperties = bindingServiceProperties.getBindingProperties("genericProcess-out-0").getProducer();
+			KafkaStreamsProducerProperties kafkaStreamsProducerProperties = kafkaStreamsExtendedBindingProperties.getExtendedProducerProperties("output");
+			Serde<?> outboundValueSerde = keyValueSerdeResolver.getOutboundValueSerde(producerProperties, kafkaStreamsProducerProperties, resolvableType.getGeneric(1));
+
+			assertThat(outboundValueSerde).isInstanceOf(GenericEventStringSerde.class);
 		}
 	}
 
@@ -105,17 +147,76 @@ public class SerdesProvidedAsBeansTests {
 		}
 	}
 
+	static class GenericEventDateSerde implements Serde<GenericEvent<Date>> {
+		@Override
+		public Serializer<GenericEvent<Date>> serializer() {
+			return null;
+		}
+
+		@Override
+		public Deserializer<GenericEvent<Date>> deserializer() {
+			return null;
+		}
+	}
+
+	static class GenericEventStringSerde implements Serde<GenericEvent<String>> {
+		@Override
+		public Serializer<GenericEvent<String>> serializer() {
+			return null;
+		}
+
+		@Override
+		public Deserializer<GenericEvent<String>> deserializer() {
+			return null;
+		}
+	}
+
+	static class GenericEvent<T> {
+
+		static <X> GenericEvent<X> of(X newThing) {
+			GenericEvent<X> newEvent = new GenericEvent<>();
+			newEvent.setThing(newThing);
+			return newEvent;
+		}
+
+		private T thing;
+
+		public T getThing() {
+			return thing;
+		}
+
+		public void setThing(T thing) {
+			this.thing = thing;
+		}
+	}
+
 	@EnableAutoConfiguration
-	public static class SerdeProvidedAsBeanApp {
+	public static class SerdesProvidedAsBeansTestApp {
 
 		@Bean
-		public Function<KStream<String, Date>, KStream<String, Date>> process() {
-			return input -> input;
+		public Serde<GenericEvent<Date>> genericEventDataSerde() {
+			return new GenericEventDateSerde();
+		}
+
+		@Bean
+		public Serde<GenericEvent<String>> genericEventStringSerde() {
+			return new GenericEventStringSerde();
 		}
 
 		@Bean
 		public Serde<Date> fooSerde() {
 			return new FooSerde<>();
 		}
+
+		@Bean
+		public Function<KStream<String, Date>, KStream<String, Date>> simpleProcess() {
+			return input -> input;
+		}
+
+		@Bean
+		public Function<KStream<String, GenericEvent<Date>>, KStream<String, GenericEvent<String>>> genericProcess() {
+			return input -> input.map((k, v) -> new KeyValue(k, GenericEvent.of(v.toString())));
+		}
+
 	}
 }