diff --git a/build.sbt b/build.sbt
index f1b9ae82c2..1aa8f5f7bd 100644
--- a/build.sbt
+++ b/build.sbt
@@ -153,7 +153,20 @@ ThisBuild / mimaBinaryIssueFilters ++= Seq(
   ProblemFilters.exclude[MissingClassProblem]("fs2.Compiler$TargetLowPriority$MonadCancelTarget"),
   ProblemFilters.exclude[MissingClassProblem]("fs2.Compiler$TargetLowPriority$MonadErrorTarget"),
   ProblemFilters.exclude[MissingTypesProblem]("fs2.Compiler$TargetLowPriority$SyncTarget"),
-  ProblemFilters.exclude[MissingClassProblem]("fs2.Chunk$VectorChunk")
+  ProblemFilters.exclude[MissingClassProblem]("fs2.Chunk$VectorChunk"),
+  ProblemFilters.exclude[ReversedMissingMethodProblem](
+    "fs2.compression.DeflateParams.fhCrcEnabled"
+  ),
+  ProblemFilters.exclude[DirectMissingMethodProblem](
+    "fs2.compression.DeflateParams#DeflateParamsImpl.copy"
+  ),
+  ProblemFilters.exclude[DirectMissingMethodProblem](
+    "fs2.compression.DeflateParams#DeflateParamsImpl.this"
+  ),
+  ProblemFilters.exclude[MissingTypesProblem]("fs2.compression.DeflateParams$DeflateParamsImpl$"),
+  ProblemFilters.exclude[DirectMissingMethodProblem](
+    "fs2.compression.DeflateParams#DeflateParamsImpl.apply"
+  )
 )
 
 lazy val root = project
diff --git a/core/jvm/src/main/scala/fs2/compression/CompressionPlatform.scala b/core/jvm/src/main/scala/fs2/compression/CompressionPlatform.scala
index abce2d4c78..f0455f01c2 100644
--- a/core/jvm/src/main/scala/fs2/compression/CompressionPlatform.scala
+++ b/core/jvm/src/main/scala/fs2/compression/CompressionPlatform.scala
@@ -404,7 +404,8 @@ private[compression] trait CompressionCompanionPlatform {
                     fileName,
                     modificationTime,
                     comment,
-                    params.level.juzDeflaterLevel
+                    params.level.juzDeflaterLevel,
+                    params.fhCrcEnabled
                   ) ++
                     _deflate(
                       params,
@@ -425,7 +426,8 @@ private[compression] trait CompressionCompanionPlatform {
           fileName: Option[String],
           modificationTime: Option[Instant],
           comment: Option[String],
-          deflateLevel: Int
+          deflateLevel: Int,
+          fhCrcEnabled: Boolean
       ): Stream[F, Byte] = {
         // See RFC 1952: https://www.ietf.org/rfc/rfc1952.txt
         val secondsSince197001010000: Long =
@@ -434,7 +436,7 @@ private[compression] trait CompressionCompanionPlatform {
           gzipMagicFirstByte, // ID1: Identification 1
           gzipMagicSecondByte, // ID2: Identification 2
           gzipCompressionMethod.DEFLATE, // CM: Compression Method
-          (gzipFlag.FHCRC + // FLG: Header CRC
+          ((if (fhCrcEnabled) gzipFlag.FHCRC else zeroByte) + // FLG: Header CRC
             fileName.map(_ => gzipFlag.FNAME).getOrElse(zeroByte) + // FLG: File name
             comment.map(_ => gzipFlag.FCOMMENT).getOrElse(zeroByte)).toByte, // FLG: Comment
           (secondsSince197001010000 & 0xff).toByte, // MTIME: Modification Time
@@ -463,10 +465,16 @@ private[compression] trait CompressionCompanionPlatform {
           bytes
         }
         val crc32Value = crc32.getValue
-        val crc16 = Array[Byte](
-          (crc32Value & 0xff).toByte,
-          ((crc32Value >> 8) & 0xff).toByte
-        )
+
+        val crc16 =
+          if (fhCrcEnabled)
+            Array[Byte](
+              (crc32Value & 0xff).toByte,
+              ((crc32Value >> 8) & 0xff).toByte
+            )
+          else
+            Array.emptyByteArray
+
         Stream.chunk(moveAsChunkBytes(header)) ++
           fileNameEncoded
             .map(bytes => Stream.chunk(moveAsChunkBytes(bytes)) ++ Stream.emit(zeroByte))
diff --git a/core/jvm/src/test/scala/fs2/JvmCompressionSuite.scala b/core/jvm/src/test/scala/fs2/JvmCompressionSuite.scala
index 29c437a675..7fa9b2eabc 100644
--- a/core/jvm/src/test/scala/fs2/JvmCompressionSuite.scala
+++ b/core/jvm/src/test/scala/fs2/JvmCompressionSuite.scala
@@ -30,6 +30,8 @@ import java.nio.charset.StandardCharsets
 import java.time.Instant
 import java.util.zip._
 import scala.collection.mutable
+import scodec.bits.crc
+import scodec.bits.ByteVector
 
 class JvmCompressionSuite extends CompressionSuite {
 
@@ -246,6 +248,35 @@ class JvmCompressionSuite extends CompressionSuite {
       .map(compressed => assert(compressed.length < uncompressed.length))
   }
 
+  test("gzip.compresses input, with FLG.FHCRC set") {
+    Stream
+      .chunk[IO, Byte](Chunk.array(getBytes("Foo")))
+      .through(
+        Compression[IO].gzip(
+          fileName = None,
+          modificationTime = None,
+          comment = None,
+          deflateParams = DeflateParams.apply(
+            bufferSize = 1024 * 32,
+            header = ZLibParams.Header.GZIP,
+            level = DeflateParams.Level.DEFAULT,
+            strategy = DeflateParams.Strategy.DEFAULT,
+            flushMode = DeflateParams.FlushMode.DEFAULT,
+            fhCrcEnabled = true
+          )
+        )
+      )
+      .compile
+      .toVector
+      .map { compressed =>
+        val headerBytes = ByteVector(compressed.take(10))
+        val crc32 = crc.crc32(headerBytes.toBitVector).toByteArray
+        val expectedCrc16 = crc32.reverse.take(2).toVector
+        val actualCrc16 = compressed.drop(10).take(2)
+        assertEquals(actualCrc16, expectedCrc16)
+      }
+  }
+
   test("gunzip limit fileName and comment length") {
     val longString: String =
       Array
diff --git a/core/shared/src/main/scala/fs2/compression/DeflateParams.scala b/core/shared/src/main/scala/fs2/compression/DeflateParams.scala
index ad6a64d63d..32c966bc29 100644
--- a/core/shared/src/main/scala/fs2/compression/DeflateParams.scala
+++ b/core/shared/src/main/scala/fs2/compression/DeflateParams.scala
@@ -45,6 +45,13 @@ sealed trait DeflateParams {
     */
   val flushMode: DeflateParams.FlushMode
 
+  /** A [[Boolean]] indicating whether the `FLG.FHCRC` bit is set. Default is `false`.
+    *  This is provided so that the compressor can be configured to have the CRC16 check enabled.
+    *  Why opt-in and not opt-out? It turned out not all clients implemented that right.
+    *  More context [[https://github.com/http4s/http4s/issues/5417 in this issue]].
+    */
+  val fhCrcEnabled: Boolean
+
   private[fs2] val bufferSizeOrMinimum: Int = bufferSize.max(128)
 }
 
@@ -57,14 +64,25 @@ object DeflateParams {
       strategy: DeflateParams.Strategy = DeflateParams.Strategy.DEFAULT,
       flushMode: DeflateParams.FlushMode = DeflateParams.FlushMode.DEFAULT
   ): DeflateParams =
-    DeflateParamsImpl(bufferSize, header, level, strategy, flushMode)
+    DeflateParamsImpl(bufferSize, header, level, strategy, flushMode, false)
+
+  def apply(
+      bufferSize: Int,
+      header: ZLibParams.Header,
+      level: DeflateParams.Level,
+      strategy: DeflateParams.Strategy,
+      flushMode: DeflateParams.FlushMode,
+      fhCrcEnabled: Boolean
+  ): DeflateParams =
+    DeflateParamsImpl(bufferSize, header, level, strategy, flushMode, fhCrcEnabled)
 
   private case class DeflateParamsImpl(
       bufferSize: Int,
       header: ZLibParams.Header,
       level: DeflateParams.Level,
       strategy: DeflateParams.Strategy,
-      flushMode: DeflateParams.FlushMode
+      flushMode: DeflateParams.FlushMode,
+      fhCrcEnabled: Boolean
   ) extends DeflateParams
 
   sealed abstract class Level(private[fs2] val juzDeflaterLevel: Int)