diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/ExtendedBindingHandlerMappingsProviderAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/ExtendedBindingHandlerMappingsProviderAutoConfiguration.java
new file mode 100644
index 000000000..a613ad555
--- /dev/null
+++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/ExtendedBindingHandlerMappingsProviderAutoConfiguration.java
@@ -0,0 +1,51 @@
+/*
+ * Copyright 2018-2021 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.HashMap;
+import java.util.Map;
+
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.context.properties.source.ConfigurationPropertyName;
+import org.springframework.cloud.stream.config.BindingHandlerAdvise.MappingsProvider;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * {@link EnableAutoConfiguration Auto-configuration} for extended binding metadata for Kafka Streams.
+ *
+ * @author Chris Bono
+ * @since 3.2
+ */
+@Configuration(proxyBeanMethods = false)
+public class ExtendedBindingHandlerMappingsProviderAutoConfiguration {
+
+	@Bean
+	public MappingsProvider kafkaStreamsExtendedPropertiesDefaultMappingsProvider() {
+		return () -> {
+			Map<ConfigurationPropertyName, ConfigurationPropertyName> mappings = new HashMap<>();
+			mappings.put(
+					ConfigurationPropertyName.of("spring.cloud.stream.kafka.streams"),
+					ConfigurationPropertyName.of("spring.cloud.stream.kafka.streams.default"));
+			mappings.put(
+					ConfigurationPropertyName.of("spring.cloud.stream.kafka.streams.bindings"),
+					ConfigurationPropertyName.of("spring.cloud.stream.kafka.streams.default"));
+			return mappings;
+		};
+	}
+
+}
diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/resources/META-INF/spring.factories b/spring-cloud-stream-binder-kafka-streams/src/main/resources/META-INF/spring.factories
index ca6492f05..00fdadf9e 100644
--- a/spring-cloud-stream-binder-kafka-streams/src/main/resources/META-INF/spring.factories
+++ b/spring-cloud-stream-binder-kafka-streams/src/main/resources/META-INF/spring.factories
@@ -1,4 +1,5 @@
 org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+  org.springframework.cloud.stream.binder.kafka.streams.ExtendedBindingHandlerMappingsProviderAutoConfiguration,\
   org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsBinderSupportAutoConfiguration,\
   org.springframework.cloud.stream.binder.kafka.streams.function.KafkaStreamsFunctionAutoConfiguration,\
   org.springframework.cloud.stream.binder.kafka.streams.endpoint.KafkaStreamsTopologyEndpointAutoConfiguration
diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/ExtendedBindingHandlerMappingsProviderAutoConfigurationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/ExtendedBindingHandlerMappingsProviderAutoConfigurationTests.java
new file mode 100644
index 000000000..43f21e243
--- /dev/null
+++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/ExtendedBindingHandlerMappingsProviderAutoConfigurationTests.java
@@ -0,0 +1,83 @@
+/*
+ * Copyright 2019-2021 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 org.junit.jupiter.api.Test;
+
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link ExtendedBindingHandlerMappingsProviderAutoConfiguration}.
+ */
+class ExtendedBindingHandlerMappingsProviderAutoConfigurationTests {
+
+	private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+			.withUserConfiguration(KafkaStreamsTestApp.class)
+			.withPropertyValues(
+					"spring.cloud.stream.kafka.streams.default.consumer.application-id: testApp123",
+					"spring.cloud.stream.kafka.streams.default.consumer.consumed-as: default-consumer",
+					"spring.cloud.stream.kafka.streams.default.consumer.materialized-as: default-materializer",
+					"spring.cloud.stream.kafka.streams.default.producer.produced-as: default-producer",
+					"spring.cloud.stream.kafka.streams.default.producer.key-serde: default-foo");
+
+	@Test
+	void defaultsUsedWhenNoCustomBindingProperties() {
+		this.contextRunner.run((context) -> {
+			assertThat(context)
+					.hasNotFailed()
+					.hasSingleBean(KafkaStreamsExtendedBindingProperties.class);
+			KafkaStreamsExtendedBindingProperties extendedBindingProperties = context.getBean(KafkaStreamsExtendedBindingProperties.class);
+			assertThat(extendedBindingProperties.getExtendedConsumerProperties("process-in-0"))
+					.hasFieldOrPropertyWithValue("applicationId", "testApp123")
+					.hasFieldOrPropertyWithValue("consumedAs", "default-consumer")
+					.hasFieldOrPropertyWithValue("materializedAs", "default-materializer");
+			assertThat(extendedBindingProperties.getExtendedProducerProperties("process-out-0"))
+					.hasFieldOrPropertyWithValue("producedAs", "default-producer")
+					.hasFieldOrPropertyWithValue("keySerde", "default-foo");
+		});
+	}
+
+	@Test
+	void defaultsRespectedWhenCustomBindingProperties() {
+		this.contextRunner
+				.withPropertyValues(
+					"spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.consumed-as: custom-consumer",
+					"spring.cloud.stream.kafka.streams.bindings.process-out-0.producer.produced-as: custom-producer")
+				.run((context) -> {
+					assertThat(context)
+							.hasNotFailed()
+							.hasSingleBean(KafkaStreamsExtendedBindingProperties.class);
+					KafkaStreamsExtendedBindingProperties extendedBindingProperties = context.getBean(KafkaStreamsExtendedBindingProperties.class);
+					assertThat(extendedBindingProperties.getExtendedConsumerProperties("process-in-0"))
+							.hasFieldOrPropertyWithValue("applicationId", "testApp123")
+							.hasFieldOrPropertyWithValue("consumedAs", "custom-consumer")
+							.hasFieldOrPropertyWithValue("materializedAs", "default-materializer");
+					assertThat(extendedBindingProperties.getExtendedProducerProperties("process-out-0"))
+							.hasFieldOrPropertyWithValue("producedAs", "custom-producer")
+							.hasFieldOrPropertyWithValue("keySerde", "default-foo");
+				});
+	}
+
+	@EnableAutoConfiguration
+	static class KafkaStreamsTestApp {
+	}
+
+}
diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ExtendedBindingHandlerMappingsProviderConfiguration.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ExtendedBindingHandlerMappingsProviderConfiguration.java
index 2c7e413ed..c05c715e3 100644
--- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ExtendedBindingHandlerMappingsProviderConfiguration.java
+++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ExtendedBindingHandlerMappingsProviderConfiguration.java
@@ -40,10 +40,6 @@ public class ExtendedBindingHandlerMappingsProviderConfiguration {
 			mappings.put(
 					ConfigurationPropertyName.of("spring.cloud.stream.kafka.bindings"),
 					ConfigurationPropertyName.of("spring.cloud.stream.kafka.default"));
-			mappings.put(
-					ConfigurationPropertyName.of("spring.cloud.stream.kafka.streams"),
-					ConfigurationPropertyName
-							.of("spring.cloud.stream.kafka.streams.default"));
 			return mappings;
 		};
 	}