diff --git a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc
index b9c2fd588..ef3f5532a 100644
--- a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc
+++ b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc
@@ -84,7 +84,6 @@ For common configuration options and properties pertaining to binder, refer to t
 === Kafka Streams Properties
 
 The following properties are available at the binder level and must be prefixed with `spring.cloud.stream.kafka.streams.binder.`
-literal.
 
 configuration::
   Map with a key/value pair containing properties pertaining to Apache Kafka Streams API.
@@ -120,7 +119,7 @@ applicationId::
 +
 Default: `none`
 
-The following properties are _only_ available for Kafka Streams producers and must be prefixed with `spring.cloud.stream.kafka.streams.bindings.<binding name>.producer.` literal.
+The following properties are _only_ available for Kafka Streams producers and must be prefixed with `spring.cloud.stream.kafka.streams.bindings.<binding name>.producer.`
 For convenience, if there multiple output bindings and they all require a common value, that can be configured by using the prefix `spring.cloud.stream.kafka.streams.default.producer.`.
 
 keySerde::
@@ -136,8 +135,8 @@ useNativeEncoding::
 +
 Default: `false`.
 
-The following properties are _only_ available for Kafka Streams consumers and must be prefixed with `spring.cloud.stream.kafka.streams.bindings.<binding name>.consumer.`literal.
-For convenience, if there multiple input bindings and they all require a common value, that can be configured by using the prefix `spring.cloud.stream.kafka.streams.default.consumer.`.
+The following properties are available for Kafka Streams consumers and must be prefixed with `spring.cloud.stream.kafka.streams.bindings.<binding-name>.consumer.`
+For convenience, if there are multiple input bindings and they all require a common value, that can be configured by using the prefix `spring.cloud.stream.kafka.streams.default.consumer.`.
 
 applicationId::
  Setting application.id per input binding.
diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java
index beded2e45..2afc35809 100644
--- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java
+++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java
@@ -443,14 +443,20 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
 		Map<String, Object> streamConfigGlobalProperties = applicationContext.getBean("streamConfigGlobalProperties", Map.class);
 
 		KafkaStreamsConsumerProperties extendedConsumerProperties = kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties(inboundName);
+		streamConfigGlobalProperties.putAll(extendedConsumerProperties.getConfiguration());
 
 		String applicationId = extendedConsumerProperties.getApplicationId();
-
 		//override application.id if set at the individual binding level.
 		if (StringUtils.hasText(applicationId)) {
 			streamConfigGlobalProperties.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
 		}
 
+		int concurrency = bindingServiceProperties.getConsumerProperties(inboundName).getConcurrency();
+		// override concurrency if set at the individual binding level.
+		if (concurrency > 1) {
+			streamConfigGlobalProperties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, concurrency);
+		}
+
 		Map<String, KafkaStreamsDlqDispatch> kafkaStreamsDlqDispatchers = applicationContext.getBean("kafkaStreamsDlqDispatchers", Map.class);
 
 		KafkaStreamsConfiguration kafkaStreamsConfiguration = new KafkaStreamsConfiguration(streamConfigGlobalProperties) {
diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java
index 841d2f921..1f091f82c 100644
--- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java
+++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderWordCountIntegrationTests.java
@@ -28,6 +28,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.streams.KafkaStreams;
 import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.Materialized;
 import org.apache.kafka.streams.kstream.Serialized;
@@ -132,6 +133,7 @@ public class KafkaStreamsBinderWordCountIntegrationTests {
 				"--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde",
 				"--spring.cloud.stream.kafka.streams.timeWindow.length=5000",
 				"--spring.cloud.stream.kafka.streams.timeWindow.advanceBy=0",
+				"--spring.cloud.stream.bindings.input.consumer.concurrency=2",
 				"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(),
 				"--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString())) {
 			receiveAndValidate(context);
@@ -141,6 +143,12 @@ public class KafkaStreamsBinderWordCountIntegrationTests {
 			ReadOnlyWindowStore<Object, Object> store = kafkaStreams.store("foo-WordCounts", QueryableStoreTypes.windowStore());
 			assertThat(store).isNotNull();
 
+			Map streamConfigGlobalProperties = context.getBean("streamConfigGlobalProperties", Map.class);
+
+			//Ensure that concurrency settings are mapped to number of stream task threads in Kafka Streams.
+			final Integer concurrency = (Integer)streamConfigGlobalProperties.get(StreamsConfig.NUM_STREAM_THREADS_CONFIG);
+			assertThat(concurrency).isEqualTo(2);
+
 			sendTombStoneRecordsAndVerifyGracefulHandling();
 
 			CleanupConfig cleanup = TestUtils.getPropertyValue(streamsBuilderFactoryBean, "cleanupConfig",