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 11bd7874d..3d73fd2f6 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
@@ -207,7 +207,7 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
 	 * @return {@link Serde} for the state store key.
 	 */
 	public Serde<?> getStateStoreKeySerde(String keySerdeString) {
-		return getKeySerde(keySerdeString, (Map<String, ?>) null);
+		return getKeySerde(keySerdeString, null);
 	}
 
 	/**
@@ -217,7 +217,7 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
 	 */
 	public Serde<?> getStateStoreValueSerde(String valueSerdeString) {
 		try {
-			return getValueSerde(valueSerdeString, (Map<String, ?>) null);
+			return getValueSerde(valueSerdeString, null);
 		}
 		catch (ClassNotFoundException ex) {
 			throw new IllegalStateException("Serde class not found: ", ex);
@@ -242,20 +242,18 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
 	}
 
 	private Serde<?> getKeySerde(String keySerdeString, ResolvableType resolvableType, Map<String, ?> extendedConfiguration) {
-		Serde<?> keySerde = null;
+		Serde<?> keySerde;
 		try {
 			if (StringUtils.hasText(keySerdeString)) {
 				keySerde = Utils.newInstance(keySerdeString, Serde.class);
 			}
 			else {
+				keySerde = Serdes.ByteArray();
 				if (resolvableType != null &&
 						(isResolvalbeKafkaStreamsType(resolvableType) || isResolvableKStreamArrayType(resolvableType))) {
-					ResolvableType generic = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(0) : resolvableType.getGeneric(0);
+					ResolvableType targetType = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(0) : resolvableType.getGeneric(0);
 					Serde<?> fallbackSerde = getFallbackSerde("default.key.serde");
-					keySerde = SerdeResolverUtils.resolveForType(this.context, generic, fallbackSerde);
-				}
-				if (keySerde == null) {
-					keySerde = Serdes.ByteArray();
+					keySerde = SerdeResolverUtils.resolveForType(this.context, targetType, fallbackSerde);
 				}
 			}
 			keySerde.configure(combineStreamConfigProperties(extendedConfiguration), true);
@@ -298,24 +296,19 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
 				: Serdes.ByteArray();
 	}
 
-	@SuppressWarnings("unchecked")
 	private Serde<?> getValueSerde(String valueSerdeString, ResolvableType resolvableType, Map<String, ?> extendedConfiguration)
 			throws ClassNotFoundException {
-		Serde<?> valueSerde = null;
+		Serde<?> valueSerde;
 		if (StringUtils.hasText(valueSerdeString)) {
 			valueSerde = Utils.newInstance(valueSerdeString, Serde.class);
 		}
 		else {
-
+			valueSerde = Serdes.ByteArray();
 			if (resolvableType != null && ((isResolvalbeKafkaStreamsType(resolvableType)) ||
 					(isResolvableKStreamArrayType(resolvableType)))) {
+				ResolvableType targetType = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(1) : resolvableType.getGeneric(1);
 				Serde<?> fallbackSerde = getFallbackSerde("default.value.serde");
-				ResolvableType generic = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(1) : resolvableType.getGeneric(1);
-				valueSerde = SerdeResolverUtils.resolveForType(this.context, generic, fallbackSerde);
-			}
-			if (valueSerde == null) {
-
-				valueSerde = Serdes.ByteArray();
+				valueSerde = SerdeResolverUtils.resolveForType(this.context, targetType, fallbackSerde);
 			}
 		}
 		valueSerde.configure(combineStreamConfigProperties(extendedConfiguration), false);
@@ -329,7 +322,7 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
 
 	private Map<String, ?> combineStreamConfigProperties(Map<String, ?> extendedConfiguration) {
 		if (extendedConfiguration != null && !extendedConfiguration.isEmpty()) {
-			Map<String, Object> streamConfiguration = new HashMap(this.streamConfigGlobalProperties);
+			Map<String, Object> streamConfiguration = new HashMap<>(this.streamConfigGlobalProperties);
 			streamConfiguration.putAll(extendedConfiguration);
 			return streamConfiguration;
 		}
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
index 3509bc29b..f03368746 100644
--- 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
@@ -16,15 +16,13 @@
 
 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.Set;
 import java.util.stream.Collectors;
 
 import org.apache.commons.logging.Log;
@@ -32,11 +30,12 @@ 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.beans.factory.BeanFactoryUtils;
+import org.springframework.beans.factory.config.BeanDefinition;
 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.lang.Nullable;
 
 /**
  * Utility class that contains various methods to help resolve {@link Serde Serdes}.
@@ -48,124 +47,120 @@ abstract class SerdeResolverUtils {
 
 	private static final Log LOG = LogFactory.getLog(SerdeResolverUtils.class);
 
+	/** Classnames of the standard built-in Serdes supported in {@link Serdes#serdeFrom(Class)}. */
+	private static final Set<String> STANDARD_SERDE_CLASSNAMES = Set.of(
+		Serdes.String().getClass().getName(),
+		Serdes.Short().getClass().getName(),
+		Serdes.Integer().getClass().getName(),
+		Serdes.Long().getClass().getName(),
+		Serdes.Float().getClass().getName(),
+		Serdes.Double().getClass().getName(),
+		Serdes.ByteArray().getClass().getName(),
+		Serdes.ByteBuffer().getClass().getName(),
+		Serdes.Bytes().getClass().getName(),
+		Serdes.UUID().getClass().getName());
+
 	/**
-	 *  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.
-	 *
+	 *  Return the {@code Serde<?>} to use for the specified type using the following steps until a match is found.
+	 *  <p><ul>
+	 *    <li>the closest matching configured {@code Serde<?>} bean if one exists</li>
+	 *    <li>the Kafka Streams built-in serde if the target type is one of the built-in types exposed by Kafka Streams
+	 *    (Integer, Long, Short, Double, Float, byte[], UUID and String)
+	 *    <li>the fallback serde if specified and not one of the Kafka Streams exposed type serdes</li>
+	 *	  <li>the {@link JsonSerde} if the target type is not exactly {@code Object}</li>
+	 *	  <li>the fallback as the last resort</li>
+	 *  </ul>
 	 * @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
+	 * @param fallbackSerde the serde to use when no type can be inferred
+	 * @return serde to use for the target type or {@code fallbackSerde} as outlined in the method description
 	 */
-	static Serde<?> resolveForType(ConfigurableApplicationContext context, ResolvableType targetType, Serde<?> fallbackSerde) {
+	static Serde<?> resolveForType(ConfigurableApplicationContext context, ResolvableType targetType, @Nullable 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();
+
+		// We don't attempt to find a matching Serde for type '?' - just return fallback
+		if (genericRawClazz == null) {
+			return fallbackSerde;
 		}
-		else if (Long.class.isAssignableFrom(genericRawClazz)) {
-			serde = Serdes.Long();
+
+		List<String> matchingSerdes = beanNamesForMatchingSerdes(context, targetType);
+		if (!matchingSerdes.isEmpty()) {
+			return context.getBean(matchingSerdes.get(0), Serde.class);
 		}
-		else if (Short.class.isAssignableFrom(genericRawClazz)) {
-			serde = Serdes.Short();
+
+		// Use standard serde for built-in types
+		Serde<?> standardDefaultSerde = getStandardDefaultSerde(genericRawClazz);
+		if (standardDefaultSerde != null) {
+			return standardDefaultSerde;
 		}
-		else if (Double.class.isAssignableFrom(genericRawClazz)) {
-			serde = Serdes.Double();
+
+		// Use fallback if specified and not from std defaults (we know from above that type is not std default
+		// so using a fallback that is std default type would not work)
+		if (fallbackSerde != null && !isSerdeFromStandardDefaults(fallbackSerde)) {
+			return fallbackSerde;
 		}
-		else if (Float.class.isAssignableFrom(genericRawClazz)) {
-			serde = Serdes.Float();
+
+		// Use JsonSerde if type is not exactly Object
+		if (!genericRawClazz.isAssignableFrom((Object.class))) {
+			return new JsonSerde<>(genericRawClazz);
 		}
-		else if (byte[].class.isAssignableFrom(genericRawClazz)) {
-			serde = Serdes.ByteArray();
+
+		// Finally, just resort to using the fallback
+		return fallbackSerde;
+	}
+
+	private static Serde<?> getStandardDefaultSerde(Class<?> genericRawClazz) {
+		try {
+			return Serdes.serdeFrom(genericRawClazz);
 		}
-		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);
+		catch (IllegalArgumentException ex) {
+			if (LOG.isTraceEnabled()) {
+				LOG.trace(ex);
 			}
 		}
-		return serde;
+		return null;
 	}
 
 	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;
-			}
+		if (serde == null) {
+			return false;
 		}
-		return false;
+		return STANDARD_SERDE_CLASSNAMES.contains(serde.getClass().getName());
 	}
 
 	/**
-	 * Find all {@link Serde} beans that are assignable from {@code targetType}.
+	 * Find the names of all {@link Serde} beans that can be used for {@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
+	 * @return list of bean names for matching serdes ordered by most specific match, or an empty list if no matches found
 	 */
-	static List<Serde<?>> findMatchingSerdes(ConfigurableApplicationContext context, ResolvableType targetType) {
+	static List<String> beanNamesForMatchingSerdes(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());
+		ResolvableType serdeType = ResolvableType.forClassWithGenerics(Serde.class, targetType);
+		String[] serdeBeanNames = BeanFactoryUtils.beanNamesForTypeIncludingAncestors(context.getBeanFactory(),
+			serdeType, false, false);
+		Arrays.stream(serdeBeanNames).forEach((beanName) -> {
 			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);
+				BeanDefinition beanDefinition = context.getBeanFactory().getMergedBeanDefinition(beanName);
+				ResolvableType serdeBeanGeneric = beanDefinition.getResolvableType().getGeneric(0);
+				if (LOG.isDebugEnabled()) {
+					LOG.debug("Found matching Serde<" + serdeBeanGeneric.getType() + "> under beanName=" + beanName);
 				}
+				matchingSerdes.add(new SerdeWithSpecificityScore(calculateScore(targetType, serdeBeanGeneric), beanName));
+			}
+			catch (Exception ex) {
+				LOG.warn("Failed introspecting Serde bean '" + beanName + "'", ex);
 			}
 		});
 		if (!matchingSerdes.isEmpty()) {
 			return matchingSerdes.stream().sorted(Collections.reverseOrder())
-				.map(SerdeWithSpecificityScore::getSerde)
+				.map(SerdeWithSpecificityScore::getSerdeBeanName)
 				.collect(Collectors.toList());
 		}
 		return Collections.emptyList();
@@ -182,7 +177,7 @@ abstract class SerdeResolverUtils {
 	 * <p><br><b>Example:</b>
 	 * <pre>{@code
 	 * -------------------------------------------------------------------------------------------------------
-	 * targetType: Foo<Date>               toString='Foo<Date>'     typeName='Foo<java.util.Date>'
+	 * 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>'
 	 * -------------------------------------------------------------------------------------------------------
@@ -221,15 +216,15 @@ abstract class SerdeResolverUtils {
 	 */
 	private static class SerdeWithSpecificityScore implements Comparable<SerdeWithSpecificityScore> {
 		private Integer score;
-		private Serde<?> serde;
+		private String serdeBeanName;
 
-		SerdeWithSpecificityScore(Integer score, Serde<?> serde) {
+		SerdeWithSpecificityScore(Integer score, String serdeBeanName) {
 			this.score = Objects.requireNonNull(score);
-			this.serde = Objects.requireNonNull(serde);
+			this.serdeBeanName = Objects.requireNonNull(serdeBeanName);
 		}
 
-		Serde<?> getSerde() {
-			return serde;
+		String getSerdeBeanName() {
+			return serdeBeanName;
 		}
 
 		@Override
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
index f59e28b50..0bf35156a 100644
--- 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
@@ -16,77 +16,349 @@
 
 package org.springframework.cloud.stream.binder.kafka.streams;
 
+import java.nio.ByteBuffer;
 import java.util.Date;
+import java.util.UUID;
+import java.util.stream.Stream;
 
 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.apache.kafka.common.utils.Bytes;
+import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 
+import org.springframework.boot.autoconfigure.AutoConfigurations;
 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 org.springframework.kafka.support.serializer.JsonSerde;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+import static org.mockito.Mockito.mock;
 
 /**
  * Unit tests for {@link SerdeResolverUtils}.
  *
  * @author Chris Bono
  */
+@SuppressWarnings({ "rawtypes", "NewClassNamingConvention", "unchecked" })
 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() {
+	@Nested
+	class ResolveForType {
 
-		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);
+		private ApplicationContextRunner contextRunner = new ApplicationContextRunner();
 
-		ApplicationContextRunner contextRunner = new ApplicationContextRunner()
-			.withUserConfiguration(SerdeResolverTestApp.class);
+		private Serde<?> fallback = mock(Serde.class);
 
-		contextRunner.run((context) -> {
+		@Test
+		void returnsFallbackSerdeForWildcard() {
+			this.contextRunner
+				.withConfiguration(AutoConfigurations.of(SerdeResolverSimpleTestApp.class))
+				.run((context) -> {
+					ResolvableType wildcardType = ResolvableType.forClass(Serde.class).getGeneric(0);
+					assertThat(SerdeResolverUtils.resolveForType(context, wildcardType, fallback)).isSameAs(fallback);
+				});
+		}
 
-			assertThat(SerdeResolverUtils.findMatchingSerdes(context, geDate))
-				.extracting("name")
-				.containsExactly("genericEventDateSerde", "genericEventDateBoundedSerde", "genericEventWildcardSerde", "genericEventRawSerde");
+		@Test
+		void returnsSerdeBeanForMatchingType() {
+			this.contextRunner
+				.withConfiguration(AutoConfigurations.of(SerdeResolverSimpleTestApp.class))
+				.run((context) -> {
+					ResolvableType fooType = ResolvableType.forClass(Foo.class);
+					assertThat(SerdeResolverUtils.resolveForType(context, fooType, fallback)).isInstanceOf(FooSerde.class);
+				});
+		}
 
-			assertThat(SerdeResolverUtils.findMatchingSerdes(context, geBounded))
-				.extracting("name")
-				.containsExactly("genericEventDateBoundedSerde", "genericEventWildcardSerde");
+		@Nested
+		class NoMatchingSerdeBeans {
 
-			assertThat(SerdeResolverUtils.findMatchingSerdes(context, geWildcard))
-				.extracting("name")
-				.containsExactly("genericEventWildcardSerde");
+			@ParameterizedTest
+			@MethodSource("kafkaStreamsBuiltInTypes")
+			void returnsStandardSerdeForKafkaStreamsBuiltInType(Class<?> builtInType, Serde<?> expectedBuiltInSerde) {
+				contextRunner.run((context) ->
+					assertThat(SerdeResolverUtils.resolveForType(context, ResolvableType.forClass(builtInType), fallback))
+						.isInstanceOf(expectedBuiltInSerde.getClass()));
+			}
 
-			// 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 Stream<Arguments> kafkaStreamsBuiltInTypes() {
+				return Stream.of(
+					arguments(String.class, Serdes.String()),
+					arguments(Short.class, Serdes.Short()),
+					arguments(Integer.class, Serdes.Integer()),
+					arguments(Long.class, Serdes.Long()),
+					arguments(Float.class, Serdes.Float()),
+					arguments(Double.class, Serdes.Double()),
+					arguments(byte[].class, Serdes.ByteArray()),
+					arguments(ByteBuffer.class, Serdes.ByteBuffer()),
+					arguments(Bytes.class, Serdes.Bytes()),
+					arguments(UUID.class, Serdes.UUID())
+				);
+			}
+
+			@Nested
+			class ForNonKafkaStreamsBuiltInType {
+
+				@Test
+				void returnsFallbackSerdeWhenValidFallbackSpecified() {
+					contextRunner.run((context) ->
+						assertThat(SerdeResolverUtils.resolveForType(context, ResolvableType.forClass(Foo.class), fallback))
+							.isSameAs(fallback));
+				}
+
+				@ParameterizedTest
+				@MethodSource("invalidFallbackSerdeProvider")
+				void returnsJsonSerdeWhenInvalidFallbackSpecified(Serde<?> invalidFallback) {
+					contextRunner.run((context) ->
+						assertThat(SerdeResolverUtils.resolveForType(context, ResolvableType.forClass(Foo.class), invalidFallback))
+							.isInstanceOf(JsonSerde.class));
+				}
+
+				static Stream<Serde<?>> invalidFallbackSerdeProvider() {
+					return Stream.of(
+						Serdes.String(),
+						Serdes.Short(),
+						Serdes.Integer(),
+						Serdes.Long(),
+						Serdes.Float(),
+						Serdes.Double(),
+						Serdes.ByteArray(),
+						Serdes.ByteBuffer(),
+						Serdes.Bytes(),
+						Serdes.UUID()
+					);
+				}
+
+				@Test
+				void returnsJsonSerdeWhenFallbackNotSpecified() {
+					contextRunner.run((context ->
+						assertThat(SerdeResolverUtils.resolveForType(context, ResolvableType.forClass(Foo.class), null))
+							.isInstanceOf(JsonSerde.class)));
+				}
+
+				@Test
+				void returnsFallbackSerdeForJavaLangObject() {
+					// This is an edge case as the only way to get to the JsonSerde step in the 1st place is when
+					// no fallback is specified or a fallback is specified but its invalid (for a built-in type).
+					// We will use the 'fallback is not specified' scenario for this test.
+					contextRunner.run((context) ->
+						assertThat(SerdeResolverUtils.resolveForType(context, ResolvableType.forClass(Object.class), null))
+							.isNull());
+				}
+			}
+		}
+	}
+
+	@Nested
+	class BeanNamesForMatchingSerdes {
+
+		@Test
+		void returnsNoSerdesForWildcardType() {
+			new ApplicationContextRunner().withUserConfiguration(SerdeResolverSimpleTestApp.class)
+				.run((context) -> {
+					ResolvableType wildcardType = ResolvableType.forClass(Serde.class).getGeneric(0);
+					assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, wildcardType)).isEmpty();
+				});
+		}
+
+		/**
+		 *
+		 * Verify that {@link SerdeResolverUtils#beanNamesForMatchingSerdes} returns the proper serdes in the proper order
+		 * for the following grid:
+		 * <p><br>
+		 * <i>NOTE:</i> {@code GE = GenericEvent}
+		 * <p><pre>{@code
+		 * ------------------------------------------------------------------
+		 * KStream type       | Serde type
+		 * ------------------------------------------------------------------
+		 *                    | GE<Date> | GE<? extends Date> | GE<?> | GE
+		 * ------------------------------------------------------------------
+		 * GE<Date>           | 1        | -                  | -     | -
+		 * GE<? extends Date> | 2        | 1                  | -     | -
+		 * GE<?>              | 4        | 2                  | 1     | 3
+		 * GE                 | 1        | -                  | -     | 2
+		 * ------------------------------------------------------------------
+		 * }</pre>
+		 * <p><br>
+		 * <i>NOTE:</i> On the last row, one might expect the {@code GE} serde to be the top match with the {@code GE}
+		 * kstream. However, that is not the case because {@code GE} is a parameterized type and therefore when
+		 * specified as a raw type (without any type info) it resolves to {@code GE<?>} which throws off the ordering.
+		 * A best practice is to specify the type info (even if it is wildcard) for KStream parameterized types.
+		 */
+		@Test
+		void returnsProperlyOrderedSerdesForSimpleGenericTypes() {
+
+			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);
+
+			new ApplicationContextRunner().withUserConfiguration(SerdeResolverSimpleTestApp.class).run((context) -> {
+
+				assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geDate))
+					.containsExactly(
+						"geDateSerde");
+
+				assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geBounded))
+					.containsExactly(
+						"geDateBoundedSerde",
+						"geDateSerde");
+
+				assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geWildcard))
+					.containsExactly(
+						"geWildcardSerde",
+						"geDateBoundedSerde",
+						"geRawSerde",
+						"geDateSerde",
+						"geStringSerde");
+
+				assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geRaw))
+					.containsExactly(
+						"geDateSerde",
+						"geStringSerde",
+						"geRawSerde");
+			});
+		}
+
+		/**
+		 * Verify that {@link SerdeResolverUtils#beanNamesForMatchingSerdes} returns the proper serdes in the proper order
+		 * for the following grid:
+		 * <p><br>
+		 * <i>NOTE:</i> {@code GE = GenericEvent}
+		 * <p><pre>{@code
+		 * -------------------------------------------------------------------------------------------------------------------
+		 * KStream type | Serde type
+		 * -------------------------------------------------------------------------------------------------------------------
+		 *              | GE<F<D>> | GE<F<?D>> | GE<F<?>> | GE<F> | GE<?F<D>> | GE<?F<?D>> | GE<?F<?>> | GE<?F> | GE<?> | GE
+		 * -------------------------------------------------------------------------------------------------------------------
+		 * GE<F<D>>     | 1        | -         | -        | -     | -         | -          | -         | -      | -     | -
+		 * GE<F<?D>>    | 2        | 1         | -        | -     | -         | -          | -         | -      | -     | -
+		 * GE<F<?>>     | 4        | 3         | 1        | 2     | -         | -          | -         | -      | -     | -
+		 * GE<F>        | 2        | -         | -        | 1     | -         | -          | -         | -      | -     | -
+		 * GE<?F<D>>    | 2        | -         | -        | -     | 1         | -          | -         | -      | -     | -
+		 * GE<?F<?D>>   | 3        | 4         | -        | -     | 2         | 1          | -         | -      | -     | -
+		 * GE<?F<?>>    | 7        | 8         | 5        | 6     | 4         | 3          | 1         | 2      | -     | -
+		 * GE<?F>       | 4        | -         | -        | 3     | 2         | -          | -         | 1      | -     | -
+		 * GE<?>        | 7        | 8         | 9        | 10    | 2         | 3          | 4         | 5      | 1     | 6
+		 * GE           | 1        | 2         | 3        | 4     | -         | -          | -         | -      | -     | 5
+		 * -------------------------------------------------------------------------------------------------------------------
+		 * }</pre>
+		 * <p><br>
+		 * <i>NOTE:</i> On the last row, one might expect the {@code GE} serde to be the top match with the {@code GE}
+		 * kstream. However, that is not the case because {@code GE} is a parameterized type and therefore when
+		 * specified as a raw type (without any type info) it resolves to {@code GE<?>} which throws off the ordering.
+		 * A best practice is to specify the type info (even if it is wildcard) for KStream parameterized types.
+		 */
+		@Test
+		void returnsProperlyOrderedSerdesForComplexGenericTypes() {
+
+			ResolvableType geFooDate = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<Foo<Date>>>() { });
+			ResolvableType geFooDateBounded = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<Foo<? extends Date>>>() { });
+			ResolvableType geFooWildcard = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<Foo<?>>>() { });
+			ResolvableType geFooRaw = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<Foo>>() { });
+			ResolvableType geFooBoundedDate = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<? extends Foo<Date>>>() { });
+			ResolvableType geFooBoundedDateBounded = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<? extends Foo<? extends Date>>>() { });
+			ResolvableType geFooBoundedWildcard = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<? extends Foo<?>>>() { });
+			ResolvableType geFooBoundedRaw = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<? extends Foo>>() { });
+			ResolvableType geWildcard = ResolvableType.forType(new ParameterizedTypeReference<GenericEvent<?>>() { });
+			ResolvableType geRaw = ResolvableType.forRawClass(GenericEvent.class);
+
+			new ApplicationContextRunner().withUserConfiguration(SerdeResolverComplexTestApp.class).run((context) -> {
+
+				assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geFooDate))
+					.containsExactly(
+						"geFooDateSerde");
+
+				assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geFooDateBounded))
+					.containsExactly(
+						"geFooDateBoundedSerde",
+						"geFooDateSerde");
+
+				assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geFooWildcard))
+					.containsExactly(
+						"geFooWildcardSerde",
+						"geFooRawSerde",
+						"geFooDateBoundedSerde",
+						"geFooDateSerde",
+						"geFooStringSerde");
+
+				assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geFooRaw))
+					.containsExactly(
+						"geFooRawSerde",
+						"geFooDateSerde",
+						"geFooStringSerde");
+
+				assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geFooBoundedDate))
+					.containsExactly(
+						"geFooBoundedDateSerde",
+						"geFooDateSerde");
+
+				assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geFooBoundedDateBounded))
+					.containsExactly(
+						"geFooBoundedDateBoundedSerde",
+						"geFooBoundedDateSerde",
+						"geFooDateSerde",
+						"geFooDateBoundedSerde");
+
+				assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geFooBoundedWildcard))
+					.containsExactly(
+						"geFooBoundedWildcardSerde",
+						"geFooBoundedRawSerde",
+						"geFooBoundedDateBoundedSerde",
+						"geFooBoundedDateSerde",
+						"geFooBoundedStringSerde",
+						"geFooWildcardSerde",
+						"geFooRawSerde",
+						"geFooDateSerde",
+						"geFooDateBoundedSerde",
+						"geFooStringSerde");
+
+				assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geFooBoundedRaw))
+					.containsExactly(
+						"geFooBoundedRawSerde",
+						"geFooBoundedDateSerde",
+						"geFooBoundedStringSerde",
+						"geFooRawSerde",
+						"geFooDateSerde",
+						"geFooStringSerde");
+
+				assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geWildcard))
+					.containsExactly(
+						"geWildcardSerde",
+						"geFooBoundedDateSerde",
+						"geFooBoundedDateBoundedSerde",
+						"geFooBoundedStringSerde",
+						"geFooBoundedWildcardSerde",
+						"geFooBoundedRawSerde",
+						"geRawSerde",
+						"geFooDateSerde",
+						"geFooDateBoundedSerde",
+						"geFooStringSerde",
+						"geFooWildcardSerde",
+						"geFooRawSerde");
+
+				// One might expect geRawSerde to win in order, but it does not because GE is a parameterized type and
+				// therefore GE resolves to GE<?> which throws off the ordering. Bottom line, for parameterized types
+				// be sure to specify a type (even if it's wildcard) in the KStream<T> definition.
+				assertThat(SerdeResolverUtils.beanNamesForMatchingSerdes(context, geRaw))
+					.containsExactly(
+						"geFooDateSerde",
+						"geFooDateBoundedSerde",
+						"geFooStringSerde",
+						"geFooWildcardSerde",
+						"geFooRawSerde",
+						"geRawSerde");
+			});
+		}
 	}
 
 	static class GenericEventSerde<T> implements Serde<GenericEvent<? extends T>> {
@@ -96,10 +368,6 @@ class SerdeResolverUtilsTests {
 			this.name = name;
 		}
 
-		String getName() {
-			return name;
-		}
-
 		@Override
 		public Serializer<GenericEvent<? extends T>> serializer() {
 			return null;
@@ -109,41 +377,141 @@ class SerdeResolverUtilsTests {
 		public Deserializer<GenericEvent<? extends T>> deserializer() {
 			return null;
 		}
+
+		@Override
+		public String toString() {
+			return "GenericEventSerde(" + name + ")";
+		}
 	}
 
 	static class GenericEvent<T> { }
 
+	static class FooSerde implements Serde<Foo> {
+
+		@Override
+		public Serializer<Foo> serializer() {
+			return null;
+		}
+
+		@Override
+		public Deserializer<Foo> deserializer() {
+			return null;
+		}
+	}
+
+	static class Foo<T> { }
+
 	@EnableAutoConfiguration
-	static class SerdeResolverTestApp {
+	static class SerdeResolverSimpleTestApp {
 
 		@Bean
-		public Serde<GenericEvent<Date>> genericEventDateSerde() {
-			return new GenericEventSerde("genericEventDateSerde");
+		public Serde<GenericEvent<Date>> geDateSerde() {
+			return new GenericEventSerde("geDateSerde");
 		}
 
 		@Bean
-		public Serde<GenericEvent<? extends Date>> genericEventDateBoundedSerde() {
-			return new GenericEventSerde("genericEventDateBoundedSerde");
+		public Serde<GenericEvent<? extends Date>> geDateBoundedSerde() {
+			return new GenericEventSerde("geDateBoundedSerde");
 		}
 
 		@Bean
-		public Serde<GenericEvent<String>> genericEventStringSerde() {
-			return new GenericEventSerde("genericEventStringSerde");
+		public Serde<GenericEvent<String>> geStringSerde() {
+			return new GenericEventSerde("geStringSerde");
 		}
 
 		@Bean
-		public Serde<GenericEvent<?>> genericEventWildcardSerde() {
-			return new GenericEventSerde("genericEventWildcardSerde");
+		public Serde<GenericEvent<?>> geWildcardSerde() {
+			return new GenericEventSerde("geWildcardSerde");
 		}
 
 		@Bean
-		public Serde<GenericEvent> genericEventRawSerde() {
-			return new GenericEventSerde("genericEventRawSerde");
+		public Serde<GenericEvent> geRawSerde() {
+			return new GenericEventSerde("geRawSerde");
 		}
 
 		@Bean
 		public Serde<?> widlcardSerde() {
 			return Serdes.Void();
 		}
+
+		@Bean
+		public Serde<Foo> fooSerde() {
+			return new FooSerde();
+		}
 	}
+
+	@EnableAutoConfiguration
+	static class SerdeResolverComplexTestApp {
+
+		@Bean
+		public Serde<GenericEvent<Foo<Date>>> geFooDateSerde() {
+			return new GenericEventSerde("geFooDateSerde");
+		}
+
+		@Bean
+		public Serde<GenericEvent<Foo<? extends Date>>> geFooDateBoundedSerde() {
+			return new GenericEventSerde("geFooDateBoundedSerde");
+		}
+
+		@Bean
+		public Serde<GenericEvent<Foo<String>>> geFooStringSerde() {
+			return new GenericEventSerde("geFooStringSerde");
+		}
+
+		@Bean
+		public Serde<GenericEvent<Foo<?>>> geFooWildcardSerde() {
+			return new GenericEventSerde("geFooWildcardSerde");
+		}
+
+		@Bean
+		public Serde<GenericEvent<Foo>> geFooRawSerde() {
+			return new GenericEventSerde("geFooRawSerde");
+		}
+
+		@Bean
+		public Serde<GenericEvent<? extends Foo<Date>>> geFooBoundedDateSerde() {
+			return new GenericEventSerde("geFooBoundedDateSerde");
+		}
+
+		@Bean
+		public Serde<GenericEvent<? extends Foo<? extends Date>>> geFooBoundedDateBoundedSerde() {
+			return new GenericEventSerde("geFooBoundedDateBoundedSerde");
+		}
+
+		@Bean
+		public Serde<GenericEvent<? extends Foo<String>>> geFooBoundedStringSerde() {
+			return new GenericEventSerde("geFooBoundedStringSerde");
+		}
+
+		@Bean
+		public Serde<GenericEvent<? extends Foo<?>>> geFooBoundedWildcardSerde() {
+			return new GenericEventSerde("geFooBoundedWildcardSerde");
+		}
+
+		@Bean
+		public Serde<GenericEvent<? extends Foo>> geFooBoundedRawSerde() {
+			return new GenericEventSerde("geFooBoundedRawSerde");
+		}
+
+		@Bean
+		public Serde<GenericEvent<?>> geWildcardSerde() {
+			return new GenericEventSerde("geWildcardSerde");
+		}
+
+		@Bean
+		public Serde<GenericEvent> geRawSerde() {
+			return new GenericEventSerde("geRawSerde");
+		}
+
+		@Bean
+		public Serde<?> widlcardSerde() {
+			return Serdes.Void();
+		}
+
+		@Bean
+		public Serde<Foo> fooSerde() {
+			return new FooSerde();
+		}
+	}
+
 }