diff --git a/codegen/src/main/scala/org/apache/pekko/grpc/gen/scaladsl/Method.scala b/codegen/src/main/scala/org/apache/pekko/grpc/gen/scaladsl/Method.scala
index 96dda0809..182efeaf4 100644
--- a/codegen/src/main/scala/org/apache/pekko/grpc/gen/scaladsl/Method.scala
+++ b/codegen/src/main/scala/org/apache/pekko/grpc/gen/scaladsl/Method.scala
@@ -66,6 +66,8 @@ case class Method(
     else if (ReservedMethodNames.contains(name)) s"$name$ReservedMethodNameSuffix"
     else name
 
+  val handlerName: String = s"${name}Handler"
+
 }
 
 object Method {
diff --git a/codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt b/codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt
index e8c6af53e..61df22b91 100644
--- a/codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt
+++ b/codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt
@@ -107,21 +107,38 @@ object @{serviceName}Handler {
           pekko.grpc.scaladsl.ServerReflection.partial(List(@{service.name})))
 }
 
-    private def methodName(request: model.HttpRequest, prefix: String): String =
+    private def methodName(request: model.HttpRequest, prefix: String): String = {
+      // Use Uri.Path structure matching instead of toString to avoid
+      // renderPath + StringBuilder + String allocations per request.
+      // gRPC path format: /{service}/{method}
       request.uri.path match {
-        case model.Uri.Path.Slash(model.Uri.Path.Segment(`prefix`, model.Uri.Path.Slash(model.Uri.Path.Segment(method, model.Uri.Path.Empty)))) =>
-          method
-        case _ =>
-          null
+        case model.Uri.Path.Slash(
+          model.Uri.Path.Segment(serviceName,
+            model.Uri.Path.Slash(
+              model.Uri.Path.Segment(method, model.Uri.Path.Empty))))
+          if serviceName == prefix => method
+        case _ => null
       }
+    }
 
     private def handler(implementation: @serviceName, prefix: String, eHandler: ActorSystem => PartialFunction[Throwable, Trailers])(implicit system: ClassicActorSystemProvider): model.HttpRequest => scala.concurrent.Future[model.HttpResponse] = {
       implicit val mat: Materializer = SystemMaterializer(system).materializer
       implicit val ec: ExecutionContext = mat.executionContext
       val spi = TelemetryExtension(system).spi
+      val compressionThreshold = system.classicSystem.settings.config.getInt("pekko.grpc.server.compression-threshold")
 
       import @{service.name}.Serializers.@{service.scalaCompatConstants.WildcardImport}
 
+      @if(!powerApis) {
+      // Pre-create lambdas for unary methods to avoid per-request allocation
+      @for(method <- service.methods) {
+      @if(method.methodType == org.apache.pekko.grpc.gen.Unary) {
+      val @{method.handlerName}: @method.parameterType => scala.concurrent.Future[@method.outputTypeUnboxed] =
+        (e: @method.parameterType) => implementation.@{method.nameSafe}(e@{if(powerApis) { ", metadata" } else { "" }})
+      }
+      }
+      }
+
       def handle(request: model.HttpRequest, method: String): scala.concurrent.Future[model.HttpResponse] =
         GrpcMarshalling.negotiated(request, (reader, writer) =>
           method match {
@@ -129,7 +146,7 @@ object @{serviceName}Handler {
             case "@method.grpcName" =>
                 @{if(powerApis) { "val metadata = MetadataBuilder.fromHeaders(request.headers)" } else { "" }}
                 @if(method.methodType == org.apache.pekko.grpc.gen.Unary) {
-                GrpcMarshalling.handleUnary(request.entity, (e: @method.parameterType) => implementation.@{method.nameSafe}(e@{if(powerApis) { ", metadata" } else { "" }}), eHandler)(
+                GrpcMarshalling.handleUnary(request.entity, @{if(powerApis) { s"(e: ${method.parameterType}) => implementation.${method.nameSafe}(e, metadata)" } else method.handlerName}, eHandler)(
                   @{service.scalaCompatConstants.ImplicitUsing}@method.deserializer.name, @method.serializer.name, mat, reader, writer, system, ec)
                 } else {
                 @{method.unmarshal}(request.entity)(@{service.scalaCompatConstants.ImplicitUsing}@method.deserializer.name, mat, reader)
@@ -140,7 +157,7 @@ object @{serviceName}Handler {
             }
             case m => GrpcExceptionHandler.from(eHandler(system.classicSystem))(system, writer)(new NotImplementedError(s"Not implemented: $m"))
           }
-      ).getOrElse(unsupportedMediaType)
+      , compressionThreshold).getOrElse(unsupportedMediaType)
 
       request => {
         val method = methodName(request, prefix)
@@ -162,9 +179,20 @@ object @{serviceName}Handler {
       implicit val mat: Materializer = SystemMaterializer(system).materializer
       implicit val ec: ExecutionContext = mat.executionContext
       val spi = TelemetryExtension(system).spi
+      val compressionThreshold = system.classicSystem.settings.config.getInt("pekko.grpc.server.compression-threshold")
 
       import @{service.name}.Serializers.@{service.scalaCompatConstants.WildcardImport}
 
+      @if(!powerApis) {
+      // Pre-create lambdas for unary methods to avoid per-request allocation
+      @for(method <- service.methods) {
+      @if(method.methodType == org.apache.pekko.grpc.gen.Unary) {
+      val @{method.handlerName}: @method.parameterType => scala.concurrent.Future[@method.outputTypeUnboxed] =
+        (e: @method.parameterType) => implementation.@{method.nameSafe}(e@{if(powerApis) { ", metadata" } else { "" }})
+      }
+      }
+      }
+
       def handle(request: model.HttpRequest, method: String): scala.concurrent.Future[model.HttpResponse] =
         GrpcMarshalling.negotiated(request, (reader, writer) =>
           method match {
@@ -172,7 +200,7 @@ object @{serviceName}Handler {
             case "@method.grpcName" =>
                 @{if(powerApis) { "val metadata = MetadataBuilder.fromHeaders(request.headers)" } else { "" }}
                 @if(method.methodType == org.apache.pekko.grpc.gen.Unary) {
-                GrpcMarshalling.handleUnary(request.entity, (e: @method.parameterType) => implementation.@{method.nameSafe}(e@{if(powerApis) { ", metadata" } else { "" }}), eHandler)(
+                GrpcMarshalling.handleUnary(request.entity, @{if(powerApis) { s"(e: ${method.parameterType}) => implementation.${method.nameSafe}(e, metadata)" } else method.handlerName}, eHandler)(
                   @{service.scalaCompatConstants.ImplicitUsing}@method.deserializer.name, @method.serializer.name, mat, reader, writer, system, ec)
                 } else {
                 @{method.unmarshal}(request.entity)(@{service.scalaCompatConstants.ImplicitUsing}@method.deserializer.name, mat, reader)
@@ -183,7 +211,7 @@ object @{serviceName}Handler {
             }
             case m => GrpcExceptionHandler.from(eHandler(system.classicSystem))(system, writer)(new NotImplementedError(s"Not implemented: $m"))
           }
-      ).getOrElse(unsupportedMediaType)
+      , compressionThreshold).getOrElse(unsupportedMediaType)
 
       Function.unlift((req: model.HttpRequest) => {
         val method = methodName(req, prefix)
diff --git a/runtime/src/main/resources/reference.conf b/runtime/src/main/resources/reference.conf
index 99fcc62a3..9085cfc23 100644
--- a/runtime/src/main/resources/reference.conf
+++ b/runtime/src/main/resources/reference.conf
@@ -1,56 +1,69 @@
 # SPDX-License-Identifier: Apache-2.0
 
 //#defaults
-pekko.grpc.client."*" {
-  # netty or pekko-http (experimental)
-  backend = "netty"
-
-  # Host to use if service-discovery-mechanism is set to static or grpc-dns
-  host = ""
-
-  service-discovery {
-    mechanism = "static"
-    # Service name to use if a service-discovery.mechanism other than static or grpc-dns
-    service-name = ""
-    # See https://pekko.apache.org/docs/pekko-management/current/discovery/index.html for meanings for each mechanism
-    # if blank then not passed to the lookup
-    port-name = ""
-    protocol = ""
-
-    # timeout for service discovery resolving
-    resolve-timeout = 1s
+pekko.grpc {
+  # Server-side configuration for gRPC services
+  server {
+    # Adaptive compression: messages smaller than this threshold (in bytes)
+    # are sent uncompressed, while larger messages are compressed with gzip.
+    # This avoids the CPU overhead of gzip for small messages where the
+    # bandwidth savings are negligible.
+    # Set to 0 to always compress (every message goes through gzip).
+    # Set to a very large value (e.g. 2147483647) to effectively disable compression.
+    compression-threshold = 1024
   }
 
-  # port to use if service-discovery-mechanism is static or service discovery does not return a port
-  port = 0
-
-  # Experimental in grpc-java https://github.com/grpc/grpc-java/issues/1771
-  # pick_first or round_robin
-  load-balancing-policy = ""
-
-  deadline = infinite
-  override-authority = ""
-  user-agent = ""
-  # Location on the classpath of CA PEM to trust
-  trusted = ""
-  use-tls = true
-  # SSL provider to use:
-  # leave empty to auto-detect, or configure 'jdk' or 'openssl'.
-  ssl-provider = ""
-
-  # TODO: Enforce HTTP/2 TLS restrictions: https://tools.ietf.org/html/draft-ietf-httpbis-http2-17#section-9.2
-
-  # The number of times to try connecting before giving up.
-  # '-1': means retry indefinitely, '0' is invalid, '1' means fail
-  # after the first failed attempt.
-  # When load balancing we don't count individual connection
-  # failures, so in that case any value larger than '1' is also
-  # interpreted as retrying 'indefinitely'.
-  connection-attempts = 20
-
-  # Service discovery mechanism to use. The default is to use a static host
-  # and port that will be resolved via DNS.
-  # Any of the mechanisms described in https://pekko.apache.org/docs/pekko-management/current/discovery/index.html can be used
-  # including Kubernetes, Consul, AWS API
+  client."*" {
+    # netty or pekko-http (experimental)
+    backend = "netty"
+
+    # Host to use if service-discovery-mechanism is set to static or grpc-dns
+    host = ""
+
+    service-discovery {
+      mechanism = "static"
+      # Service name to use if a service-discovery.mechanism other than static or grpc-dns
+      service-name = ""
+      # See https://pekko.apache.org/docs/pekko-management/current/discovery/index.html for meanings for each mechanism
+      # if blank then not passed to the lookup
+      port-name = ""
+      protocol = ""
+
+      # timeout for service discovery resolving
+      resolve-timeout = 1s
+    }
+
+    # port to use if service-discovery-mechanism is static or service discovery does not return a port
+    port = 0
+
+    # Experimental in grpc-java https://github.com/grpc/grpc-java/issues/1771
+    # pick_first or round_robin
+    load-balancing-policy = ""
+
+    deadline = infinite
+    override-authority = ""
+    user-agent = ""
+    # Location on the classpath of CA PEM to trust
+    trusted = ""
+    use-tls = true
+    # SSL provider to use:
+    # leave empty to auto-detect, or configure 'jdk' or 'openssl'.
+    ssl-provider = ""
+
+    # TODO: Enforce HTTP/2 TLS restrictions: https://tools.ietf.org/html/draft-ietf-httpbis-http2-17#section-9.2
+
+    # The number of times to try connecting before giving up.
+    # '-1': means retry indefinitely, '0' is invalid, '1' means fail
+    # after the first failed attempt.
+    # When load balancing we don't count individual connection
+    # failures, so in that case any value larger than '1' is also
+    # interpreted as retrying 'indefinitely'.
+    connection-attempts = 20
+
+    # Service discovery mechanism to use. The default is to use a static host
+    # and port that will be resolved via DNS.
+    # Any of the mechanisms described in https://pekko.apache.org/docs/pekko-management/current/discovery/index.html can be used
+    # including Kubernetes, Consul, AWS API
+  }
 }
 //#defaults
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala b/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala
index 1d8e2f042..4bf7da081 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala
@@ -18,7 +18,16 @@ import pekko.NotUsed
 import pekko.annotation.InternalApi
 import pekko.annotation.InternalStableApi
 import pekko.grpc.GrpcProtocol.{ GrpcProtocolReader, GrpcProtocolWriter }
-import pekko.grpc.internal.{ Codec, Codecs, GrpcProtocolNative, GrpcProtocolWeb, GrpcProtocolWebText }
+import pekko.grpc.internal.{
+  AdaptiveGzip,
+  Codec,
+  Codecs,
+  GrpcProtocolNative,
+  GrpcProtocolWeb,
+  GrpcProtocolWebText,
+  Gzip,
+  Identity
+}
 import pekko.http.javadsl.{ model => jmodel }
 import pekko.http.scaladsl.model.{ ContentType, HttpHeader, HttpResponse, Trailer }
 import pekko.http.scaladsl.model.HttpEntity.ChunkStreamPart
@@ -141,15 +150,99 @@ object GrpcProtocol {
     case _                                       => None
   }
 
+  // Pre-computed negotiation result for Identity reader + Identity writer (no gzip support)
+  private val NativeIdentityIdentity: (Try[GrpcProtocolReader], GrpcProtocolWriter) =
+    (scala.util.Success(GrpcProtocolNative.newReader(Identity)), GrpcProtocolNative.newWriter(Identity))
+
+  // Pre-computed negotiation result for Identity reader + AdaptiveGzip writer (default threshold)
+  // This is the most common case: client accepts gzip, server uses adaptive compression.
+  private val NativeIdentityAdaptiveGzip: (Try[GrpcProtocolReader], GrpcProtocolWriter) =
+    (scala.util.Success(GrpcProtocolNative.newReader(Identity)),
+      GrpcProtocolNative.newWriter(AdaptiveGzip(AdaptiveGzip.DefaultCompressionThreshold)))
+
   /**
    * Calculates the gRPC protocol encoding to use for an interaction with a gRPC client.
+   * Optimized with a fast path for native gRPC that does a single header scan.
+   *
+   * For response encoding, uses adaptive gzip: messages above a configurable threshold
+   * (default 1KB) are compressed with gzip, while smaller messages are sent uncompressed.
+   * The gRPC per-frame compression flag (bit 0 of the 5-byte frame header) tells the
+   * client whether each individual frame is compressed.
    *
    * @param request the client request to respond to.
    * @return the protocol reader for the request, and a protocol writer for the response.
    */
   def negotiate(request: jmodel.HttpRequest): Option[(Try[GrpcProtocolReader], GrpcProtocolWriter)] =
-    detect(request).map { variant =>
-      (Codecs.detect(request).map(variant.newReader), variant.newWriter(Codecs.negotiate(request)))
+    negotiate(request, AdaptiveGzip.DefaultCompressionThreshold)
+
+  /**
+   * Calculates the gRPC protocol encoding to use for an interaction with a gRPC client.
+   * Optimized with a fast path for native gRPC that does a single header scan.
+   *
+   * @param request the client request to respond to.
+   * @param compressionThreshold minimum message size in bytes to trigger gzip compression.
+   *                             Messages below this threshold are sent uncompressed.
+   * @return the protocol reader for the request, and a protocol writer for the response.
+   */
+  def negotiate(
+      request: jmodel.HttpRequest,
+      compressionThreshold: Int): Option[(Try[GrpcProtocolReader], GrpcProtocolWriter)] = {
+    val mediaType = request.entity.getContentType.mediaType
+    val subType = mediaType.subType
+    val isNative = (subType == "grpc+proto") || (subType == "grpc")
+
+    if (isNative) {
+      // Single-pass header scan for native gRPC
+      var requestEncoding: String = null
+      var acceptEncoding: String = null
+      request match {
+        case sReq: pekko.http.scaladsl.model.HttpMessage =>
+          val headers = sReq.headers
+          var i = 0
+          while (i < headers.size) {
+            val h = headers(i)
+            val name = h.lowercaseName
+            if (name == "grpc-encoding") requestEncoding = h.value
+            else if (name == "grpc-accept-encoding") acceptEncoding = h.value
+            i += 1
+          }
+        case _ =>
+      }
+      // Determine reader codec (request encoding)
+      val readerCodec: Codec =
+        if (requestEncoding eq null) Identity
+        else if (requestEncoding == "identity") Identity
+        else if (requestEncoding == "gzip") Gzip
+        else return slowNegotiateOpt(request, subType)
+      // Determine writer codec (response encoding)
+      // Adaptive compression: use gzip for messages above threshold, identity for smaller ones.
+      // Only activate when client advertises gzip support via grpc-accept-encoding.
+      val clientAcceptsGzip = acceptEncoding != null && acceptEncoding.contains("gzip")
+      if (!clientAcceptsGzip && (readerCodec eq Identity)) return Some(NativeIdentityIdentity)
+      if (clientAcceptsGzip && (readerCodec eq Identity) &&
+        compressionThreshold == AdaptiveGzip.DefaultCompressionThreshold)
+        return Some(NativeIdentityAdaptiveGzip)
+      val writerCodec: Codec =
+        if (clientAcceptsGzip) AdaptiveGzip(compressionThreshold)
+        else Identity
+      return Some(
+        (scala.util.Success(GrpcProtocolNative.newReader(readerCodec)),
+          GrpcProtocolNative.newWriter(writerCodec)))
+    }
+
+    // Non-native protocols - slow path
+    slowNegotiateOpt(request, subType)
+  }
+
+  private def slowNegotiateOpt(request: jmodel.HttpRequest, subType: String)
+      : Option[(Try[GrpcProtocolReader], GrpcProtocolWriter)] = {
+    val variant = subType match {
+      case "grpc-web" | "grpc-web+proto"           => Some(GrpcProtocolWeb)
+      case "grpc-web-text" | "grpc-web-text+proto" => Some(GrpcProtocolWebText)
+      case "grpc" | "grpc+proto"                   => Some(GrpcProtocolNative)
+      case _                                       => None
     }
+    variant.map(v => (Codecs.detect(request).map(v.newReader), v.newWriter(Codecs.negotiate(request))))
+  }
 
 }
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala b/runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala
index e5392960e..148032839 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/ProtobufSerializer.scala
@@ -27,4 +27,22 @@ trait ProtobufSerializer[T] {
 
 private[grpc] trait ProtobufFrameSerializer[T] extends ProtobufSerializer[T] {
   private[grpc] def serializeDataFrame(t: T): ByteString
+
+  /**
+   * Deserialize a protobuf message from a ByteString starting at the given offset.
+   * Avoids ByteString.slice allocation by passing the offset directly to the parser.
+   * Default implementation falls back to slice + deserialize.
+   */
+  private[grpc] def deserialize(data: ByteString, offset: Int, length: Int): T =
+    deserialize(data.slice(offset, offset + length))
+
+  /**
+   * Returns the serialized size of the message without actually serializing it.
+   * Used by adaptive compression to decide whether to compress small messages.
+   *
+   * Implementations should override this with an efficient method (e.g.
+   * `GeneratedMessage.serializedSize` for ScalaPB or `Message.getSerializedSize`
+   * for Google protobuf) to avoid the overhead of full serialization.
+   */
+  private[grpc] def serializedDataSize(t: T): Int = serialize(t).length
 }
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala b/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
index 8cd4653cb..2a11b7078 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
@@ -100,10 +100,19 @@ object AbstractGrpcProtocol {
       })
       .toContentType
 
+  private val SmallMessageThreshold = 4096
+
   def encodeFrameData(data: ByteString, isCompressed: Boolean, isTrailer: Boolean): ByteString = {
-    val header = new Array[Byte](FrameHeaderSize)
-    writeFrameHeader(header, 0, data.length, isCompressed, isTrailer)
-    ByteString.fromArrayUnsafe(header, 0, FrameHeaderSize) ++ data
+    if (data.length <= SmallMessageThreshold) {
+      val combined = new Array[Byte](FrameHeaderSize + data.length)
+      writeFrameHeader(combined, 0, data.length, isCompressed, isTrailer)
+      data.copyToArray(combined, FrameHeaderSize)
+      ByteString.fromArrayUnsafe(combined)
+    } else {
+      val header = new Array[Byte](FrameHeaderSize)
+      writeFrameHeader(header, 0, data.length, isCompressed, isTrailer)
+      ByteString.fromArrayUnsafe(header, 0, FrameHeaderSize) ++ data
+    }
   }
 
   def writer(
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/AdaptiveGzip.scala b/runtime/src/main/scala/org/apache/pekko/grpc/internal/AdaptiveGzip.scala
new file mode 100644
index 000000000..48e3b5c3a
--- /dev/null
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/AdaptiveGzip.scala
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ *    http://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.apache.pekko.grpc.internal
+
+import java.io.ByteArrayOutputStream
+import java.util.zip.GZIPOutputStream
+
+import org.apache.pekko.annotation.InternalApi
+import org.apache.pekko.util.ByteString
+
+/**
+ * Adaptive gzip codec that only compresses messages above a size threshold.
+ *
+ * For small messages, the CPU overhead of gzip compression exceeds the bandwidth
+ * savings. This codec checks the message size before compressing and passes through
+ * uncompressed data for small messages, while still advertising "gzip" as the
+ * encoding to the client. The gRPC per-frame compression flag (bit 0 of the
+ * 5-byte frame header) tells the client whether each individual frame is compressed.
+ *
+ * @param compressionThreshold minimum message size in bytes to trigger compression
+ */
+@InternalApi
+private[grpc] class AdaptiveGzip(val compressionThreshold: Int) extends Codec {
+  override val name: String = "gzip"
+
+  override def compress(uncompressed: ByteString): ByteString = {
+    if (uncompressed.size < compressionThreshold) uncompressed
+    else {
+      val baos = new ByteArrayOutputStream(uncompressed.size)
+      val gzos = new GZIPOutputStream(baos)
+      try gzos.write(uncompressed.toArrayUnsafe())
+      finally gzos.close()
+      ByteString.fromArrayUnsafe(baos.toByteArray)
+    }
+  }
+
+  override def uncompress(compressed: ByteString): ByteString = Gzip.uncompress(compressed)
+
+  override def uncompress(compressedBitSet: Boolean, bytes: ByteString): ByteString =
+    Gzip.uncompress(compressedBitSet, bytes)
+
+  override def isCompressed: Boolean = true
+
+  override def compressWithFlag(bytes: ByteString): (ByteString, Boolean) = {
+    if (bytes.size < compressionThreshold) (bytes, false)
+    else {
+      val baos = new ByteArrayOutputStream(bytes.size)
+      val gzos = new GZIPOutputStream(baos)
+      try gzos.write(bytes.toArrayUnsafe())
+      finally gzos.close()
+      (ByteString.fromArrayUnsafe(baos.toByteArray), true)
+    }
+  }
+}
+
+@InternalApi
+private[grpc] object AdaptiveGzip {
+  val DefaultCompressionThreshold: Int = 1024
+
+  private val DefaultInstance: AdaptiveGzip = new AdaptiveGzip(DefaultCompressionThreshold)
+
+  def apply(threshold: Int = DefaultCompressionThreshold): AdaptiveGzip =
+    if (threshold == DefaultCompressionThreshold) DefaultInstance
+    else new AdaptiveGzip(threshold)
+}
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codec.scala b/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codec.scala
index 24b31a8f8..d4953947e 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codec.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codec.scala
@@ -28,4 +28,14 @@ abstract class Codec {
   def uncompress(compressedBitSet: Boolean, bytes: ByteString): ByteString
 
   def isCompressed: Boolean = this != Identity
+
+  /**
+   * Compress data and report whether compression was actually applied.
+   * Returns a tuple of (compressedData, wasCompressed) where wasCompressed
+   * indicates whether the data should be marked with the compression flag
+   * in the gRPC frame header. Default implementation uses compress() + isCompressed.
+   * Adaptive codecs override this to signal per-frame compression decisions.
+   */
+  def compressWithFlag(bytes: ByteString): (ByteString, Boolean) =
+    (compress(bytes), isCompressed)
 }
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codecs.scala b/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codecs.scala
index 6dfbd3577..d2876f8b2 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codecs.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codecs.scala
@@ -23,8 +23,10 @@ import scala.collection.immutable
 import scala.util.{ Failure, Success, Try }
 
 object Codecs {
-  // TODO should this list be made user-extensible?
-  val supportedCodecs = immutable.Seq(Gzip, Identity)
+  // Identity preferred over Gzip: for typical small protobuf messages, compression
+  // adds CPU overhead without meaningful size reduction. Clients requiring compression
+  // can still negotiate it explicitly.
+  val supportedCodecs = immutable.Seq(Identity, Gzip)
   private val supportedByName: Map[String, Codec] = supportedCodecs.map(c => c.name -> c).toMap
 
   private def extractHeaders(request: jm.HttpMessage): Iterable[jm.HttpHeader] = {
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala
index c7f13ff7f..031a8a797 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala
@@ -44,16 +44,33 @@ object GrpcProtocolNative extends AbstractGrpcProtocol("grpc") {
     AbstractGrpcProtocol.writer(this, codec, encodeFrame(codec, _), encodeDataToResponse(codec))
 
   override protected def reader(codec: Codec): GrpcProtocolReader =
-    AbstractGrpcProtocol.reader(codec, decodeFrame)
+    if (codec eq Identity)
+      AbstractGrpcProtocol.reader(codec, decodeFrame).copy(decodeSingleFrame = decodeIdentitySingleFrame)
+    else AbstractGrpcProtocol.reader(codec, decodeFrame)
 
   @inline
   private def decodeFrame(@nowarn("msg=is never used") frameType: Int, data: ByteString) = DataFrame(data)
 
+  private def decodeIdentitySingleFrame(frame: ByteString): ByteString = {
+    if (frame.length < AbstractGrpcProtocol.FrameHeaderSize) throw new MissingParameterException
+
+    val frameType = frame(0)
+    val length = frame.readIntBE(1)
+    val available = frame.length - AbstractGrpcProtocol.FrameHeaderSize
+    if (length > available) throw new MissingParameterException
+    if (length < 0 || length < available)
+      throw new IllegalStateException("Unexpected data")
+    if ((frameType & 0x80) != 0) throw new IllegalStateException("Cannot read unknown frame")
+
+    Identity.uncompress((frameType & 1) == 1, frame.slice(AbstractGrpcProtocol.FrameHeaderSize, frame.length))
+  }
+
   @inline
   private def encodeFrame(codec: Codec, frame: Frame): ChunkStreamPart =
     frame match {
       case DataFrame(data) =>
-        Chunk(AbstractGrpcProtocol.encodeFrameData(codec.compress(data), codec.isCompressed, isTrailer = false))
+        val (compressed, wasCompressed) = codec.compressWithFlag(data)
+        Chunk(AbstractGrpcProtocol.encodeFrameData(compressed, wasCompressed, isTrailer = false))
       case TrailerFrame(headers) => LastChunk(trailer = headers)
     }
   private def encodeDataToResponse(
@@ -65,6 +82,8 @@ object GrpcProtocolNative extends AbstractGrpcProtocol("grpc") {
       protocol = HttpProtocols.`HTTP/1.1`,
       attributes = Map.empty[AttributeKey[?], Any].updated(AttributeKeys.trailer, trailer))
 
-  private def encodeDataToFrameBytes(codec: Codec, data: ByteString): ByteString =
-    AbstractGrpcProtocol.encodeFrameData(codec.compress(data), codec.isCompressed, isTrailer = false)
+  private def encodeDataToFrameBytes(codec: Codec, data: ByteString): ByteString = {
+    val (compressed, wasCompressed) = codec.compressWithFlag(data)
+    AbstractGrpcProtocol.encodeFrameData(compressed, wasCompressed, isTrailer = false)
+  }
 }
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcResponseHelpers.scala b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcResponseHelpers.scala
index 3e285b73b..93de4052a 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcResponseHelpers.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcResponseHelpers.scala
@@ -72,10 +72,23 @@ object GrpcResponseHelpers {
       system: ClassicActorSystemProvider): HttpResponse = {
     val responseHeaders = responseHeadersFor(writer)
     try {
-      if ((writer.messageEncoding eq Identity) && writer.contentType == GrpcProtocolNative.contentType) {
-        m match {
-          case frameSerializer: ProtobufFrameSerializer[T @unchecked] =>
-            nativeResponse(writer, frameSerializer.serializeDataFrame(e), responseHeaders)
+      if (writer.contentType == GrpcProtocolNative.contentType) {
+        writer.messageEncoding match {
+          case Identity =>
+            m match {
+              case frameSerializer: ProtobufFrameSerializer[T @unchecked] =>
+                nativeResponse(writer, frameSerializer.serializeDataFrame(e), responseHeaders)
+              case _ =>
+                writer.encodeDataToResponse(m.serialize(e), responseHeaders, TrailerOkAttribute)
+            }
+          case adaptive: AdaptiveGzip =>
+            m match {
+              case frameSerializer: ProtobufFrameSerializer[T @unchecked]
+                  if frameSerializer.serializedDataSize(e) < adaptive.compressionThreshold =>
+                nativeResponse(writer, frameSerializer.serializeDataFrame(e), responseHeaders)
+              case _ =>
+                writer.encodeDataToResponse(m.serialize(e), responseHeaders, TrailerOkAttribute)
+            }
           case _ =>
             writer.encodeDataToResponse(m.serialize(e), responseHeaders, TrailerOkAttribute)
         }
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GoogleProtobufSerializer.scala b/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GoogleProtobufSerializer.scala
index c71b6ed1c..ec35b1239 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GoogleProtobufSerializer.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GoogleProtobufSerializer.scala
@@ -28,6 +28,7 @@ class GoogleProtobufSerializer[T <: com.google.protobuf.Message](parser: Parser[
 
   override def serialize(t: T): ByteString =
     ByteString.fromArrayUnsafe(t.toByteArray)
+  override private[grpc] def serializedDataSize(t: T): Int = t.getSerializedSize
   override private[grpc] def serializeDataFrame(t: T): ByteString = {
     val dataLength = t.getSerializedSize
     val frame = new Array[Byte](AbstractGrpcProtocol.FrameHeaderSize + dataLength)
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GrpcMarshalling.scala b/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GrpcMarshalling.scala
index 18c496371..2f9e896f9 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GrpcMarshalling.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GrpcMarshalling.scala
@@ -60,7 +60,15 @@ object GrpcMarshalling {
       reader: GrpcProtocolReader): CompletionStage[T] =
     entity match {
       case strict: pekko.http.scaladsl.model.HttpEntity.Strict =>
-        completedOrFailed(u.deserialize(reader.decodeSingleFrame(strict.data)))
+        completedOrFailed {
+          u match {
+            case frameSerializer: ProtobufFrameSerializer[T @unchecked] =>
+              frameSerializer.deserialize(strict.data, AbstractGrpcProtocol.FrameHeaderSize,
+                strict.data.length - AbstractGrpcProtocol.FrameHeaderSize)
+            case _ =>
+              u.deserialize(reader.decodeSingleFrame(strict.data))
+          }
+        }
       case _ =>
         unmarshal(entity.getDataBytes, u, mat, reader)
     }
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala
index 42b248fe6..65042f127 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala
@@ -55,9 +55,25 @@ object GrpcMarshalling {
   }
 
   def negotiated[T](req: HttpRequest, f: (GrpcProtocolReader, GrpcProtocolWriter) => Future[T]): Option[Future[T]] =
-    GrpcProtocol.negotiate(req).map {
-      case (Success(reader), writer) => f(reader, writer)
-      case (Failure(ex), _)          => Future.failed(ex)
+    GrpcProtocol.negotiate(req) match {
+      case Some((Success(reader), writer)) => Some(f(reader, writer))
+      case Some((Failure(ex), _))          => Some(Future.failed(ex))
+      case None                            => None
+    }
+
+  /**
+   * Like `negotiated` but accepts a pre-read compression threshold to avoid
+   * per-request config reading. The threshold should be read once at handler
+   * creation time from `pekko.grpc.server.compression-threshold`.
+   */
+  def negotiated[T](
+      req: HttpRequest,
+      f: (GrpcProtocolReader, GrpcProtocolWriter) => Future[T],
+      compressionThreshold: Int): Option[Future[T]] =
+    GrpcProtocol.negotiate(req, compressionThreshold) match {
+      case Some((Success(reader), writer)) => Some(f(reader, writer))
+      case Some((Failure(ex), _))          => Some(Future.failed(ex))
+      case None                            => None
     }
 
   def unmarshal[T](data: Source[ByteString, Any])(
@@ -125,18 +141,28 @@ object GrpcMarshalling {
     entity match {
       case HttpEntity.Strict(_, data) =>
         try {
-          val in = u.deserialize(reader.decodeSingleFrame(data))
+          val in = u match {
+            case frameSerializer: ProtobufFrameSerializer[In @unchecked] =>
+              if (data.length < AbstractGrpcProtocol.FrameHeaderSize)
+                throw new MissingParameterException
+              frameSerializer.deserialize(data, AbstractGrpcProtocol.FrameHeaderSize,
+                data.length - AbstractGrpcProtocol.FrameHeaderSize)
+            case _ =>
+              u.deserialize(reader.decodeSingleFrame(data))
+          }
           invokeUnary(in, implementation, eHandler)
         } catch {
-          case NonFatal(ex) => unaryExceptionHandler(eHandler)(system, writer)(ex)
+          case ex: MissingParameterException => handleUnaryException(ex, eHandler)
+          case NonFatal(ex)                  => handleUnaryException(ex, eHandler)
         }
       case _ =>
         val requestFuture = unmarshal[In](entity)(u, mat, reader)
         requestFuture.value match {
           case Some(Success(in)) => invokeUnary(in, implementation, eHandler)
-          case Some(Failure(ex)) => unaryExceptionHandler(eHandler)(system, writer)(ex)
+          case Some(Failure(ex)) => handleUnaryException(ex, eHandler)
           case None              =>
-            val exceptionHandler = unaryExceptionHandler(eHandler)
+            // Lazy: only create the PartialFunction if actually needed for recovery
+            lazy val exceptionHandler = unaryExceptionHandler(eHandler)
             requestFuture
               .flatMap(in => invokeUnary(in, implementation, eHandler))
               .recoverWith(exceptionHandler)
@@ -154,7 +180,7 @@ object GrpcMarshalling {
       ec: ExecutionContext): Future[HttpResponse] =
     try handleUnaryResponse(implementation(in), eHandler)
     catch {
-      case NonFatal(ex) => unaryExceptionHandler(eHandler)(system, writer)(ex)
+      case NonFatal(ex) => handleUnaryException(ex, eHandler)
     }
 
   @inline private def handleUnaryResponse[Out](
@@ -166,9 +192,9 @@ object GrpcMarshalling {
       ec: ExecutionContext): Future[HttpResponse] =
     responseFuture.value match {
       case Some(Success(out)) => marshalUnaryResponse(out, eHandler)
-      case Some(Failure(ex))  => unaryExceptionHandler(eHandler)(system, writer)(ex)
+      case Some(Failure(ex))  => handleUnaryException(ex, eHandler)
       case None               =>
-        val exceptionHandler = unaryExceptionHandler(eHandler)
+        lazy val exceptionHandler = unaryExceptionHandler(eHandler)
         responseFuture.map(out => marshal[Out](out, eHandler)(m, writer, system)).recoverWith(exceptionHandler)
     }
 
@@ -180,9 +206,25 @@ object GrpcMarshalling {
       system: ClassicActorSystemProvider): Future[HttpResponse] =
     try FastFuture.successful(marshal[Out](out, eHandler)(m, writer, system))
     catch {
-      case NonFatal(ex) => unaryExceptionHandler(eHandler)(system, writer)(ex)
+      case NonFatal(ex) => handleUnaryException(ex, eHandler)
     }
 
+  /**
+   * Inline exception handling for synchronous error paths.
+   * Avoids allocating a PartialFunction via GrpcExceptionHandler.from() for each exception.
+   * Instead, directly applies the mapper and converts to response.
+   */
+  @inline private def handleUnaryException(
+      ex: Throwable,
+      eHandler: ActorSystem => PartialFunction[Throwable, Trailers])(
+      implicit writer: GrpcProtocolWriter,
+      system: ClassicActorSystemProvider): Future[HttpResponse] = {
+    val mapper = eHandler(system.classicSystem)
+    val trailers = if (mapper.isDefinedAt(ex)) mapper(ex)
+    else GrpcExceptionHandler.defaultMapper(system.classicSystem)(ex)
+    FastFuture.successful(GrpcResponseHelpers.status(trailers))
+  }
+
   @inline private def unaryExceptionHandler(eHandler: ActorSystem => PartialFunction[Throwable, Trailers])(
       implicit system: ClassicActorSystemProvider,
       writer: GrpcProtocolWriter): PartialFunction[Throwable, Future[HttpResponse]] =
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializer.scala b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializer.scala
index a9884852c..8aa589c00 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializer.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/ScalapbProtobufSerializer.scala
@@ -29,6 +29,7 @@ class ScalapbProtobufSerializer[T <: GeneratedMessage](companion: GeneratedMessa
     extends ProtobufFrameSerializer[T] {
   override def serialize(t: T): ByteString =
     ByteString.fromArrayUnsafe(t.toByteArray)
+  override private[grpc] def serializedDataSize(t: T): Int = t.serializedSize
   override private[grpc] def serializeDataFrame(t: T): ByteString = {
     val dataLength = t.serializedSize
     val frame = new Array[Byte](AbstractGrpcProtocol.FrameHeaderSize + dataLength)
@@ -40,8 +41,25 @@ class ScalapbProtobufSerializer[T <: GeneratedMessage](companion: GeneratedMessa
 
     ByteString.fromArrayUnsafe(frame)
   }
-  override def deserialize(bytes: ByteString): T =
-    companion.parseFrom(CodedInputStream.newInstance(bytes.asByteBuffer))
+  override def deserialize(bytes: ByteString): T = {
+    // Fast path: use byte array directly to avoid ByteBuffer wrapper allocation
+    val bb = bytes.asByteBuffer
+    if (bb.hasArray) {
+      companion.parseFrom(CodedInputStream.newInstance(bb.array(), bb.arrayOffset(), bb.remaining()))
+    } else {
+      companion.parseFrom(CodedInputStream.newInstance(bb))
+    }
+  }
+  override private[grpc] def deserialize(data: ByteString, offset: Int, length: Int): T = {
+    // Offset-aware fast path: bypass ByteString.slice to avoid ByteString1 + ByteBuffer wrappers
+    val bb = data.asByteBuffer
+    if (bb.hasArray) {
+      companion.parseFrom(CodedInputStream.newInstance(bb.array(), bb.arrayOffset() + offset, length))
+    } else {
+      // Fallback for non-array-backed ByteStrings (rare in practice)
+      super.deserialize(data, offset, length)
+    }
+  }
   override def deserialize(data: InputStream): T =
     companion.parseFrom(data)
 }
diff --git a/runtime/src/test/scala/org/apache/pekko/grpc/internal/AdaptiveGzipSpec.scala b/runtime/src/test/scala/org/apache/pekko/grpc/internal/AdaptiveGzipSpec.scala
new file mode 100644
index 000000000..8e7ed5589
--- /dev/null
+++ b/runtime/src/test/scala/org/apache/pekko/grpc/internal/AdaptiveGzipSpec.scala
@@ -0,0 +1,270 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ *    http://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.apache.pekko.grpc.internal
+
+import org.apache.pekko
+import pekko.grpc.GrpcProtocol
+import pekko.grpc.scaladsl.headers
+import pekko.http.scaladsl.model.{ ContentTypes, HttpHeader, HttpRequest }
+import pekko.util.ByteString
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+import scala.collection.immutable
+
+class AdaptiveGzipSpec extends AnyWordSpec with Matchers {
+
+  private val smallData = ByteString.fromArrayUnsafe(new Array[Byte](100))
+  private val largeData = ByteString.fromArrayUnsafe(new Array[Byte](4096))
+
+  private def nativeRequest(acceptEncoding: String = null, encoding: String = null): HttpRequest = {
+    var hdrs: immutable.Seq[HttpHeader] = immutable.Seq.empty
+    if (acceptEncoding != null)
+      hdrs = hdrs :+ headers.`Message-Accept-Encoding`(acceptEncoding)
+    if (encoding != null)
+      hdrs = hdrs :+ headers.`Message-Encoding`(encoding)
+    HttpRequest(
+      headers = hdrs,
+      entity = pekko.http.scaladsl.model.HttpEntity(ContentTypes.`application/grpc+proto`, ByteString.empty))
+  }
+
+  "AdaptiveGzip" should {
+
+    "not compress data below threshold" in {
+      val codec = AdaptiveGzip(1024)
+      val result = codec.compress(smallData)
+      result should be theSameInstanceAs smallData
+    }
+
+    "compress data above threshold" in {
+      val codec = AdaptiveGzip(1024)
+      val result = codec.compress(largeData)
+      result should not be theSameInstanceAs(largeData)
+      result.length should be < largeData.length
+    }
+
+    "use default threshold of 1024" in {
+      val codec = AdaptiveGzip()
+      codec.compressionThreshold should be(1024)
+    }
+
+    "report name as gzip" in {
+      AdaptiveGzip().name should be("gzip")
+    }
+
+    "report isCompressed as true" in {
+      AdaptiveGzip().isCompressed should be(true)
+    }
+
+    "uncompress data compressed by Gzip" in {
+      val codec = AdaptiveGzip(1024)
+      val compressed = Gzip.compress(largeData)
+      val uncompressed = codec.uncompress(compressed)
+      uncompressed should be(largeData)
+    }
+
+    "uncompress with compressedBitSet=true" in {
+      val codec = AdaptiveGzip(1024)
+      val compressed = Gzip.compress(largeData)
+      val uncompressed = codec.uncompress(compressedBitSet = true, compressed)
+      uncompressed should be(largeData)
+    }
+
+    "pass through with compressedBitSet=false" in {
+      val codec = AdaptiveGzip(1024)
+      val uncompressed = codec.uncompress(compressedBitSet = false, smallData)
+      uncompressed should be theSameInstanceAs smallData
+    }
+
+    "round-trip: compress then uncompress preserves data" in {
+      val codec = AdaptiveGzip(1024)
+      val data = ByteString("Hello, World! " * 200)
+      val compressed = codec.compress(data)
+      val uncompressed = codec.uncompress(compressedBitSet = true, compressed)
+      uncompressed should be(data)
+    }
+
+    "round-trip: pass-through for small data" in {
+      val codec = AdaptiveGzip(1024)
+      val data = ByteString("small")
+      val result = codec.compress(data)
+      result should be theSameInstanceAs data
+      codec.uncompress(compressedBitSet = false, result) should be(data)
+    }
+
+    "not compress empty data" in {
+      val codec = AdaptiveGzip(1024)
+      val data = ByteString.empty
+      val result = codec.compress(data)
+      result should be theSameInstanceAs data
+    }
+
+    "always compress when threshold is 0" in {
+      val codec = AdaptiveGzip(0)
+      val data = ByteString("tiny")
+      val result = codec.compress(data)
+      result should not be theSameInstanceAs(data)
+      codec.uncompress(compressedBitSet = true, result) should be(data)
+    }
+
+    "never compress when threshold is Int.MaxValue" in {
+      val codec = AdaptiveGzip(Int.MaxValue)
+      val result = codec.compress(largeData)
+      result should be theSameInstanceAs largeData
+    }
+
+    "compress data at exactly the threshold (boundary)" in {
+      val codec = AdaptiveGzip(1024)
+      val data = ByteString.fromArrayUnsafe(new Array[Byte](1024))
+      val result = codec.compress(data)
+      result should not be theSameInstanceAs(data)
+    }
+
+    "not compress data at threshold minus 1 (boundary)" in {
+      val codec = AdaptiveGzip(1024)
+      val data = ByteString.fromArrayUnsafe(new Array[Byte](1023))
+      val result = codec.compress(data)
+      result should be theSameInstanceAs data
+    }
+
+    "compressWithFlag returns false for small data" in {
+      val codec = AdaptiveGzip(1024)
+      val (data, flag) = codec.compressWithFlag(smallData)
+      data should be theSameInstanceAs smallData
+      flag should be(false)
+    }
+
+    "compressWithFlag returns true for large data" in {
+      val codec = AdaptiveGzip(1024)
+      val (data, flag) = codec.compressWithFlag(largeData)
+      data should not be theSameInstanceAs(largeData)
+      flag should be(true)
+    }
+
+    "cache default instance" in {
+      val a = AdaptiveGzip()
+      val b = AdaptiveGzip()
+      a should be theSameInstanceAs b
+    }
+
+    "not cache non-default instances" in {
+      val a = AdaptiveGzip(2048)
+      val b = AdaptiveGzip(2048)
+      a should not be theSameInstanceAs(b)
+    }
+  }
+
+  "GrpcProtocol.negotiate" should {
+
+    "use AdaptiveGzip when client accepts gzip" in {
+      val result = GrpcProtocol.negotiate(nativeRequest(acceptEncoding = "gzip"))
+      result shouldBe defined
+      val (_, writer) = result.get
+      writer.messageEncoding shouldBe an[AdaptiveGzip]
+      writer.messageEncoding.name should be("gzip")
+    }
+
+    "use AdaptiveGzip when client accepts gzip among multiple encodings" in {
+      val result = GrpcProtocol.negotiate(nativeRequest(acceptEncoding = "identity,gzip"))
+      result shouldBe defined
+      val (_, writer) = result.get
+      writer.messageEncoding shouldBe an[AdaptiveGzip]
+    }
+
+    "use Identity when client does not accept gzip" in {
+      val result = GrpcProtocol.negotiate(nativeRequest(acceptEncoding = "identity"))
+      result shouldBe defined
+      val (_, writer) = result.get
+      writer.messageEncoding should be(Identity)
+    }
+
+    "use Identity when no accept-encoding header" in {
+      val result = GrpcProtocol.negotiate(nativeRequest())
+      result shouldBe defined
+      val (_, writer) = result.get
+      writer.messageEncoding should be(Identity)
+    }
+
+    "use Gzip reader when request has gzip encoding" in {
+      val result = GrpcProtocol.negotiate(nativeRequest(acceptEncoding = "gzip", encoding = "gzip"))
+      result shouldBe defined
+      val (readerTry, writer) = result.get
+      val reader = readerTry.get
+      reader.messageEncoding should be(Gzip)
+      writer.messageEncoding shouldBe an[AdaptiveGzip]
+    }
+
+    "use Identity reader when request has no encoding" in {
+      val result = GrpcProtocol.negotiate(nativeRequest(acceptEncoding = "gzip"))
+      result shouldBe defined
+      val (readerTry, _) = result.get
+      val reader = readerTry.get
+      reader.messageEncoding should be(Identity)
+    }
+
+    "use custom compression threshold when specified" in {
+      val result = GrpcProtocol.negotiate(nativeRequest(acceptEncoding = "gzip"), 2048)
+      result shouldBe defined
+      val (_, writer) = result.get
+      writer.messageEncoding shouldBe an[AdaptiveGzip]
+      writer.messageEncoding.asInstanceOf[AdaptiveGzip].compressionThreshold should be(2048)
+    }
+
+    "use default threshold of 1024 when no threshold specified" in {
+      val result = GrpcProtocol.negotiate(nativeRequest(acceptEncoding = "gzip"))
+      result shouldBe defined
+      val (_, writer) = result.get
+      writer.messageEncoding.asInstanceOf[AdaptiveGzip].compressionThreshold should be(1024)
+    }
+  }
+
+  "Streaming frame encoding with AdaptiveGzip" should {
+
+    "set compression flag to 0 for small messages (below threshold)" in {
+      val codec = AdaptiveGzip(1024)
+      val smallMsg = ByteString.fromArrayUnsafe(new Array[Byte](100))
+      val writer = GrpcProtocolNative.newWriter(codec)
+      val frame = writer.encodeFrame(GrpcProtocol.DataFrame(smallMsg))
+      val chunk = frame.asInstanceOf[pekko.http.scaladsl.model.HttpEntity.Chunk]
+      val frameData = chunk.data
+      frameData(0) should be(0.toByte)
+      frameData.length should be(AbstractGrpcProtocol.FrameHeaderSize + smallMsg.length)
+    }
+
+    "set compression flag to 1 for large messages (above threshold)" in {
+      val codec = AdaptiveGzip(1024)
+      val largeMsg = ByteString.fromArrayUnsafe(new Array[Byte](4096))
+      val writer = GrpcProtocolNative.newWriter(codec)
+      val frame = writer.encodeFrame(GrpcProtocol.DataFrame(largeMsg))
+      val chunk = frame.asInstanceOf[pekko.http.scaladsl.model.HttpEntity.Chunk]
+      val frameData = chunk.data
+      frameData(0) should be(1.toByte)
+    }
+
+    "round-trip small message through encode then decode" in {
+      val codec = AdaptiveGzip(1024)
+      val writer = GrpcProtocolNative.newWriter(codec)
+      val reader = GrpcProtocolNative.newReader(Identity)
+      val originalMsg = ByteString("Hello, adaptive compression!")
+      val frame = writer.encodeFrame(GrpcProtocol.DataFrame(originalMsg))
+      val chunk = frame.asInstanceOf[pekko.http.scaladsl.model.HttpEntity.Chunk]
+      val decoded = reader.decodeSingleFrame(chunk.data)
+      decoded should be(originalMsg)
+    }
+  }
+}
diff --git a/runtime/src/test/scala/org/apache/pekko/grpc/javadsl/GrpcMarshallingSpec.scala b/runtime/src/test/scala/org/apache/pekko/grpc/javadsl/GrpcMarshallingSpec.scala
index c35c50f40..aa10204cf 100644
--- a/runtime/src/test/scala/org/apache/pekko/grpc/javadsl/GrpcMarshallingSpec.scala
+++ b/runtime/src/test/scala/org/apache/pekko/grpc/javadsl/GrpcMarshallingSpec.scala
@@ -23,14 +23,16 @@ import scala.concurrent.Await
 import scala.concurrent.duration._
 
 import com.google.protobuf.{ Any => ProtobufAny, ByteString => ProtobufByteString }
+import io.grpc.StatusException
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.wordspec.AnyWordSpec
 
 import org.apache.pekko
 import pekko.actor.ActorSystem
-import pekko.grpc.internal.{ AbstractGrpcProtocol, GrpcProtocolNative, Identity }
+import pekko.grpc.internal.{ AbstractGrpcProtocol, GrpcProtocolNative, Identity, MissingParameterException }
 import pekko.http.scaladsl.model.HttpEntity
 import pekko.stream.SystemMaterializer
+import pekko.util.ByteString
 
 class GrpcMarshallingSpec extends AnyWordSpec with Matchers {
   "The javadsl GrpcMarshalling" should {
@@ -59,6 +61,45 @@ class GrpcMarshallingSpec extends AnyWordSpec with Matchers {
       }
     }
 
+    "decode a strict native identity frame directly" in {
+      val serializer = new GoogleProtobufSerializer(ProtobufAny.parser())
+      val message =
+        ProtobufAny.newBuilder().setTypeUrl("benchmark").setValue(ProtobufByteString.copyFromUtf8("payload")).build()
+      val payload = serializer.serialize(message)
+      val frame = AbstractGrpcProtocol.encodeFrameData(payload, isCompressed = false, isTrailer = false)
+
+      GrpcProtocolNative.newReader(Identity).decodeSingleFrame(frame) should be(payload)
+    }
+
+    "reject compressed strict native identity frames" in {
+      val frame = AbstractGrpcProtocol.encodeFrameData(ByteString(1, 2, 3), isCompressed = true, isTrailer = false)
+
+      a[StatusException] should be thrownBy GrpcProtocolNative.newReader(Identity).decodeSingleFrame(frame)
+    }
+
+    "reject truncated strict native identity frames" in {
+      a[MissingParameterException] should be thrownBy
+      GrpcProtocolNative
+        .newReader(Identity)
+        .decodeSingleFrame(ByteString(0, 0, 0, 0))
+    }
+
+    "reject strict native identity frames with missing payload bytes" in {
+      a[MissingParameterException] should be thrownBy
+      GrpcProtocolNative
+        .newReader(Identity)
+        .decodeSingleFrame(ByteString(0, 0, 0, 0, 3, 1, 2))
+    }
+
+    "reject strict native identity frames with trailing bytes" in {
+      val frame = AbstractGrpcProtocol.encodeFrameData(ByteString(1, 2, 3), isCompressed = false, isTrailer = false)
+
+      an[IllegalStateException] should be thrownBy
+      GrpcProtocolNative
+        .newReader(Identity)
+        .decodeSingleFrame(frame ++ ByteString(4))
+    }
+
     "recover a failed unary response stage" in {
       val system = ActorSystem("GrpcMarshallingSpec")
       try {