diff --git a/README.adoc b/README.adoc
index 9758642ee..226842310 100644
--- a/README.adoc
+++ b/README.adoc
@@ -419,7 +419,7 @@ If a topic already exists with a larger number of partitions than the maximum of
 
 compressionType::
 Set the `compression.type` producer property.
-Supported values are `none`, `gzip`, `snappy` and `lz4`.
+Supported values are `none`, `gzip`, `snappy`, `lz4`, `zstd`.
 If you override the `kafka-clients` jar to 2.1.0 (or later), as discussed in the https://docs.spring.io/spring-kafka/docs/2.2.x/reference/html/deps-for-21x.html[Spring for Apache Kafka documentation], and wish to use `zstd` compression, use `spring.cloud.stream.kafka.bindings.<binding-name>.producer.configuration.compression.type=zstd`.
 +
 Default: `none`.
diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc
index 16db230a9..a9bbb7ff5 100644
--- a/docs/src/main/asciidoc/overview.adoc
+++ b/docs/src/main/asciidoc/overview.adoc
@@ -398,7 +398,7 @@ If a topic already exists with a larger number of partitions than the maximum of
 
 compression::
 Set the `compression.type` producer property.
-Supported values are `none`, `gzip`, `snappy` and `lz4`.
+Supported values are `none`, `gzip`, `snappy`, `lz4` and `zstd`.
 If you override the `kafka-clients` jar to 2.1.0 (or later), as discussed in the https://docs.spring.io/spring-kafka/docs/2.2.x/reference/html/deps-for-21x.html[Spring for Apache Kafka documentation], and wish to use `zstd` compression, use `spring.cloud.stream.kafka.bindings.<binding-name>.producer.configuration.compression.type=zstd`.
 +
 Default: `none`.
diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java
index 3689e067a..5d0ff6eb7 100644
--- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java
+++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java
@@ -121,7 +121,7 @@ public class KafkaProducerProperties {
 	/**
 	 * @return compression type {@link CompressionType}
 	 *
-	 * Set the compression.type producer property. Supported values are none, gzip, snappy and lz4.
+	 * Set the compression.type producer property. Supported values are none, gzip, snappy, lz4 and zstd.
 	 * See {@link CompressionType} for more details.
 	 */
 	@NotNull
@@ -304,11 +304,10 @@ public class KafkaProducerProperties {
 		 */
 		lz4,
 
-		// /** // TODO: uncomment and fix docs when kafka-clients 2.1.0 or newer is the
-		// default
-		// * zstd compression
-		// */
-		// zstd
+		/**
+		 * zstd compression.
+		 */
+		zstd,
 
 	}
 
diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
index 4a77decd4..974f2adb3 100644
--- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
+++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
@@ -1430,7 +1430,8 @@ public class KafkaBinderTests extends
 		final KafkaProducerProperties.CompressionType[] codecs = new KafkaProducerProperties.CompressionType[] {
 				KafkaProducerProperties.CompressionType.none,
 				KafkaProducerProperties.CompressionType.gzip,
-				KafkaProducerProperties.CompressionType.snappy };
+				KafkaProducerProperties.CompressionType.snappy,
+				KafkaProducerProperties.CompressionType.zstd};
 		byte[] testPayload = new byte[2048];
 		Arrays.fill(testPayload, (byte) 65);
 		Binder binder = getBinder();