diff --git a/http-bench-jmh/src/main/scala/org/apache/pekko/http/shaded/com/twitter/hpack/HpackEncoderBenchmark.scala b/http-bench-jmh/src/main/scala/org/apache/pekko/http/shaded/com/twitter/hpack/HpackEncoderBenchmark.scala
new file mode 100644
index 000000000..0ad3bb9a8
--- /dev/null
+++ b/http-bench-jmh/src/main/scala/org/apache/pekko/http/shaded/com/twitter/hpack/HpackEncoderBenchmark.scala
@@ -0,0 +1,74 @@
+/*
+ * 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.http.shaded.com.twitter.hpack
+
+import java.util.concurrent.TimeUnit
+
+import org.openjdk.jmh.annotations._
+
+import org.apache.pekko.http.impl.util.ByteStringOutputStream
+
+@State(Scope.Benchmark)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@BenchmarkMode(Array(Mode.AverageTime))
+@Fork(2)
+@Warmup(iterations = 5, time = 1)
+@Measurement(iterations = 8, time = 2)
+class HpackEncoderBenchmark {
+
+  /** `default` lets the encoder pick Huffman when it is shorter, the other two force one string literal form. */
+  @Param(Array("default", "huffman", "raw"))
+  var mode = "default"
+
+  // indexing is disabled so that every call encodes the values as string literals instead of table references
+  var encoder: Encoder = null
+  val out = new ByteStringOutputStream(128)
+
+  // a typical browser request; every value goes through HuffmanEncoder.getEncodedLength and most through encode
+  val headers: Array[(String, String)] = Array(
+    ":method" -> "GET",
+    ":scheme" -> "https",
+    ":authority" -> "www.example.com",
+    ":path" -> "/api/v1/users/12345/orders?page=2&sort=created_at&direction=desc",
+    "user-agent" ->
+    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
+    "accept" -> "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
+    "accept-language" -> "en-US,en;q=0.9",
+    "accept-encoding" -> "gzip, deflate, br",
+    "cookie" -> "session=ab12cd34ef56ab12cd34ef56ab12cd34ef56ab12; theme=dark; consent=1",
+    "cache-control" -> "no-cache")
+
+  @Setup
+  def setup(): Unit =
+    encoder = mode match {
+      case "default" => new Encoder(4096, false, false, false)
+      case "huffman" => new Encoder(4096, false, true, false)
+      case "raw"     => new Encoder(4096, false, false, true)
+    }
+
+  @Benchmark
+  def encodeHeaders(): Int = {
+    var i = 0
+    while (i < headers.length) {
+      val (name, value) = headers(i)
+      encoder.encodeHeader(out, name, value, false)
+      i += 1
+    }
+    out.takeByteString().length
+  }
+}
diff --git a/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/Encoder.java b/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/Encoder.java
index c94d6df90..1bf7062eb 100644
--- a/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/Encoder.java
+++ b/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/Encoder.java
@@ -31,9 +31,8 @@
 
 package org.apache.pekko.http.shaded.com.twitter.hpack;
 
-import java.io.IOException;
-import java.io.OutputStream;
 import java.util.Arrays;
+import org.apache.pekko.http.impl.util.ByteStringOutputStream;
 import org.apache.pekko.http.impl.util.StringTools;
 import org.apache.pekko.http.shaded.com.twitter.hpack.HpackUtil.IndexType;
 
@@ -75,8 +74,8 @@ public Encoder(int maxHeaderTableSize) {
   }
 
   /** Encode the header field into the header block. */
-  public void encodeHeader(OutputStream out, String name, String value, boolean sensitive)
-      throws IOException {
+  public void encodeHeader(
+      ByteStringOutputStream out, String name, String value, boolean sensitive) {
 
     // If the header value is sensitive then it must never be indexed
     if (sensitive) {
@@ -130,7 +129,7 @@ public void encodeHeader(OutputStream out, String name, String value, boolean se
   }
 
   /** Set the maximum table size. */
-  public void setMaxHeaderTableSize(OutputStream out, int maxHeaderTableSize) throws IOException {
+  public void setMaxHeaderTableSize(ByteStringOutputStream out, int maxHeaderTableSize) {
     if (maxHeaderTableSize < 0) {
       throw new IllegalArgumentException("Illegal Capacity: " + maxHeaderTableSize);
     }
@@ -148,7 +147,7 @@ public int getMaxHeaderTableSize() {
   }
 
   /** Encode integer according to Section 5.1. */
-  private static void encodeInteger(OutputStream out, int mask, int n, int i) throws IOException {
+  private static void encodeInteger(ByteStringOutputStream out, int mask, int n, int i) {
     if (n < 0 || n > 8) {
       throw new IllegalArgumentException("N: " + n);
     }
@@ -171,23 +170,25 @@ private static void encodeInteger(OutputStream out, int mask, int n, int i) thro
   }
 
   /** Encode string literal according to Section 5.2. */
-  private void encodeStringLiteral(OutputStream out, String string) throws IOException {
-    int length = string.length();
-    int huffmanLength = Huffman.ENCODER.getEncodedLength(string);
+  private void encodeStringLiteral(ByteStringOutputStream out, String string) {
+    // convert once up front: the length computation, the Huffman coder and the raw literal all work
+    // on the octets
+    byte[] stringBytes = StringTools.asciiStringBytes(string);
+    int length = stringBytes.length;
+    int huffmanLength = Huffman.ENCODER.getEncodedLength(stringBytes);
     if ((huffmanLength < length && !forceHuffmanOff) || forceHuffmanOn) {
       encodeInteger(out, 0x80, 7, huffmanLength);
-      Huffman.ENCODER.encode(out, string);
+      int position = out.reserve(huffmanLength);
+      Huffman.ENCODER.encode(stringBytes, out.array(), position, huffmanLength);
     } else {
-      byte[] stringBytes = StringTools.asciiStringBytes(string);
       encodeInteger(out, 0x00, 7, length);
-      out.write(stringBytes, 0, stringBytes.length);
+      out.write(stringBytes, 0, length);
     }
   }
 
   /** Encode literal header field according to Section 6.2. */
   private void encodeLiteral(
-      OutputStream out, String name, String value, IndexType indexType, int nameIndex)
-      throws IOException {
+      ByteStringOutputStream out, String name, String value, IndexType indexType, int nameIndex) {
     int mask;
     int prefixBits;
     switch (indexType) {
@@ -228,7 +229,7 @@ private int getNameIndex(String name) {
    * Ensure that the dynamic table has enough room to hold 'headerSize' more bytes. Removes the
    * oldest entry from the dynamic table until sufficient space is available.
    */
-  private void ensureCapacity(int headerSize) throws IOException {
+  private void ensureCapacity(int headerSize) {
     while (size + headerSize > capacity) {
       int index = length();
       if (index == 0) {
diff --git a/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/HuffmanEncoder.java b/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/HuffmanEncoder.java
index 7b8771e8e..413c7affc 100644
--- a/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/HuffmanEncoder.java
+++ b/http-core/src/main/java/org/apache/pekko/http/shaded/com/twitter/hpack/HuffmanEncoder.java
@@ -31,9 +31,6 @@
 
 package org.apache.pekko.http.shaded.com.twitter.hpack;
 
-import java.io.IOException;
-import java.io.OutputStream;
-
 final class HuffmanEncoder {
 
   private final int[] codes;
@@ -51,25 +48,37 @@ final class HuffmanEncoder {
   }
 
   /**
-   * Compresses the input string literal using the Huffman coding.
+   * Compresses the input string literal using the Huffman coding, writing the result directly into
+   * <code>dst</code> instead of byte by byte to an <code>OutputStream</code>.
    *
-   * @param out the output stream for the compressed data
-   * @throws IOException if an I/O error occurs. In particular, an <code>IOException</code> may be
-   *     thrown if the output stream has been closed.
+   * @param data the string literal to be Huffman encoded
+   * @param dst the array to write the Huffman coded string literal to
+   * @param dstOffset the position in <code>dst</code> to start writing at
+   * @param encodedLength the value of {@link #getEncodedLength(byte[])} for <code>data</code>,
+   *     which the caller has typically already computed to decide whether to use Huffman coding and
+   *     to reserve room in <code>dst</code>
    */
-  public void encode(OutputStream out, String string) throws IOException {
-    if (out == null) {
-      throw new NullPointerException("out");
-    } else if (string == null) {
-      throw new NullPointerException("string");
+  public void encode(byte[] data, byte[] dst, int dstOffset, int encodedLength) {
+    if (data == null) {
+      throw new NullPointerException("data");
+    } else if (dst == null) {
+      throw new NullPointerException("dst");
+    } else if (dstOffset < 0 || encodedLength < 0 || dstOffset + encodedLength > dst.length) {
+      throw new IndexOutOfBoundsException(
+          "dstOffset "
+              + dstOffset
+              + ", encodedLength "
+              + encodedLength
+              + ", dst.length "
+              + dst.length);
     }
 
+    int pos = dstOffset;
     long current = 0;
     int n = 0;
-    int len = string.length();
 
-    for (int i = 0; i < len; i++) {
-      int b = string.charAt(i) & 0xFF;
+    for (byte value : data) {
+      int b = value & 0xFF;
       int code = codes[b];
       int nbits = lengths[b];
 
@@ -79,14 +88,22 @@ public void encode(OutputStream out, String string) throws IOException {
 
       while (n >= 8) {
         n -= 8;
-        out.write(((int) (current >> n)));
+        dst[pos++] = (byte) (current >> n);
       }
     }
 
     if (n > 0) {
       current <<= (8 - n);
       current |= (0xFF >>> n); // this should be EOS symbol
-      out.write((int) current);
+      dst[pos++] = (byte) current;
+    }
+
+    if (pos - dstOffset != encodedLength) {
+      throw new IllegalArgumentException(
+          "encodedLength "
+              + encodedLength
+              + " does not match the Huffman encoded length "
+              + (pos - dstOffset));
     }
   }
 
@@ -96,13 +113,13 @@ public void encode(OutputStream out, String string) throws IOException {
    * @param data the string literal to be Huffman encoded
    * @return the number of bytes required to Huffman encode <code>data</code>
    */
-  public int getEncodedLength(String data) {
+  public int getEncodedLength(byte[] data) {
     if (data == null) {
       throw new NullPointerException("data");
     }
     long len = 0;
-    for (int i = 0; i < data.length(); i++) {
-      len += lengths[data.charAt(i) & 0xFF];
+    for (byte b : data) {
+      len += lengths[b & 0xFF];
     }
     return (int) ((len + 7) >> 3);
   }
diff --git a/http-core/src/main/mima-filters/2.0.x.backwards.excludes/hpack-encoder-bytestring-output-stream.excludes b/http-core/src/main/mima-filters/2.0.x.backwards.excludes/hpack-encoder-bytestring-output-stream.excludes
new file mode 100644
index 000000000..d96eda20e
--- /dev/null
+++ b/http-core/src/main/mima-filters/2.0.x.backwards.excludes/hpack-encoder-bytestring-output-stream.excludes
@@ -0,0 +1,24 @@
+# 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.
+
+# The shaded HPACK Encoder writes into the internal ByteStringOutputStream instead of any java.io.OutputStream,
+# and the package-private HuffmanEncoder codes byte arrays instead of Strings, straight into that stream's array
+ProblemFilters.exclude[IncompatibleMethTypeProblem]("org.apache.pekko.http.shaded.com.twitter.hpack.Encoder.encodeHeader")
+ProblemFilters.exclude[IncompatibleMethTypeProblem]("org.apache.pekko.http.shaded.com.twitter.hpack.Encoder.setMaxHeaderTableSize")
+ProblemFilters.exclude[IncompatibleMethTypeProblem]("org.apache.pekko.http.shaded.com.twitter.hpack.HuffmanEncoder.encode")
+ProblemFilters.exclude[IncompatibleMethTypeProblem]("org.apache.pekko.http.shaded.com.twitter.hpack.HuffmanEncoder.getEncodedLength")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.http.shaded.com.twitter.hpack.HuffmanEncoder.encode")
diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/hpack/HeaderCompression.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/hpack/HeaderCompression.scala
index 9bce63b8e..c86ef6367 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/hpack/HeaderCompression.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/hpack/HeaderCompression.scala
@@ -13,11 +13,11 @@
 
 package org.apache.pekko.http.impl.engine.http2.hpack
 
-import java.io.ByteArrayOutputStream
 import org.apache.pekko
 import pekko.annotation.InternalApi
 import pekko.http.impl.engine.http2.Http2Protocol.SettingIdentifier
 import pekko.http.impl.engine.http2._
+import pekko.http.impl.util.ByteStringOutputStream
 import pekko.stream.{ Attributes, FlowShape, Inlet, Outlet }
 import pekko.stream.stage.{ GraphStage, GraphStageLogic, InHandler, OutHandler, StageLogging }
 import pekko.util.ByteString
@@ -44,7 +44,7 @@ private[http2] object HeaderCompression extends GraphStage[FlowShape[FrameEvent,
       private val currentMaxFrameSize = Http2Protocol.InitialMaxFrameSize
 
       val encoder = new pekko.http.shaded.com.twitter.hpack.Encoder(Http2Protocol.InitialMaxHeaderTableSize)
-      val os = new ByteArrayOutputStream(128)
+      val os = new ByteStringOutputStream(128)
 
       def onPull(): Unit = pull(eventsIn)
       def onPush(): Unit = grab(eventsIn) match {
@@ -71,8 +71,7 @@ private[http2] object HeaderCompression extends GraphStage[FlowShape[FrameEvent,
                 throw new IllegalStateException(
                   s"Didn't expect key-value-pair [$key] -> [$value](${value.getClass}) here.")
             }
-            val result = ByteString.fromArrayUnsafe(os.toByteArray) // BAOS.toByteArray always creates a copy
-            os.reset()
+            val result = os.takeByteString() // hands the array over without copying and starts a new block
             if (result.size <= currentMaxFrameSize)
               push(eventsOut, HeadersFrame(streamId, endStream, endHeaders = true, result, prioInfo))
             else {
diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/PerMessageDeflate.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/PerMessageDeflate.scala
index 8453fcc2c..24b4e2c11 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/PerMessageDeflate.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/PerMessageDeflate.scala
@@ -255,7 +255,7 @@ private[http] object PerMessageDeflate {
           output.write(buffer, 0, count)
           count = inflater.inflate(buffer)
         }
-        output.toByteStringUnsafe
+        output.takeByteString()
       } catch {
         case ex: DataFormatException =>
           throw new ProtocolException(s"Invalid WebSocket compressed message: ${ex.getMessage}")
@@ -351,7 +351,7 @@ private[http] object PerMessageDeflate {
         output.write(buffer, 0, count)
         count = deflater.deflate(buffer, 0, buffer.length, Deflater.SYNC_FLUSH)
       }
-      val bytes = output.toByteStringUnsafe
+      val bytes = output.takeByteString()
       if (removeTail && bytes.endsWith(EmptyStoredBlock)) bytes.dropRight(EmptyStoredBlock.length) else bytes
     }
 
diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/util/ByteStringOutputStream.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/util/ByteStringOutputStream.scala
index acb034ffd..5d2e77a6a 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/impl/util/ByteStringOutputStream.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/impl/util/ByteStringOutputStream.scala
@@ -17,7 +17,8 @@
 
 package org.apache.pekko.http.impl.util
 
-import java.io.ByteArrayOutputStream
+import java.io.OutputStream
+import java.util.Arrays
 
 import org.apache.pekko
 import pekko.annotation.InternalApi
@@ -27,25 +28,88 @@ import pekko.util.ByteString
  * INTERNAL API
  *
  * An [[java.io.OutputStream]] that buffers into a byte array like [[java.io.ByteArrayOutputStream]] but
- * that can hand the buffered data over as a [[pekko.util.ByteString]] without copying it, unlike
- * `ByteArrayOutputStream.toByteArray` which always creates a copy.
+ * that hands the buffered data over as a [[pekko.util.ByteString]] without copying it, unlike
+ * `ByteArrayOutputStream.toByteArray` which always creates a copy, and whose writes are not
+ * `synchronized`: every user is a single stage that owns the stream, and the per-call monitor was a
+ * measurable share of the cost of encoding an HPACK header block.
+ *
+ * After a hand-over the stream is empty again and can be reused for the next block of data.
+ *
+ * Not thread-safe.
  *
  * Derived from the `ByteStringOutputStream` in Apache Pekko gRPC
  * (https://github.com/apache/pekko-grpc/pull/862).
  */
 @InternalApi
-private[http] final class ByteStringOutputStream(capacity: Int) extends ByteArrayOutputStream(capacity) {
+private[http] final class ByteStringOutputStream(initialCapacity: Int) extends OutputStream {
+  if (initialCapacity < 0) throw new IllegalArgumentException(s"Illegal initial capacity: $initialCapacity")
+
+  private[this] var buf: Array[Byte] = Array.emptyByteArray
+  private[this] var count: Int = 0
+  // the size of the last block handed over; the array for the next block is allocated to hold it
+  private[this] var lastCount: Int = 0
+
+  /** The number of bytes written since the last hand-over. */
+  def size: Int = count
+
+  override def write(b: Int): Unit = {
+    ensureCapacity(1)
+    buf(count) = b.toByte
+    count += 1
+  }
+
+  override def write(bytes: Array[Byte], offset: Int, length: Int): Unit = {
+    ensureCapacity(length)
+    System.arraycopy(bytes, offset, buf, count, length)
+    count += length
+  }
 
   /**
-   * Wraps the bytes written so far in a `ByteString`. The buffer may be shared with the returned
-   * `ByteString`, so this stream must not be written to, reset or reused afterwards.
+   * Reserves `length` bytes at the end of the buffered data for the caller to fill in directly.
+   *
+   * @return the position in [[array]] at which the reserved bytes start
    */
-  def toByteStringUnsafe: ByteString =
+  def reserve(length: Int): Int = {
+    ensureCapacity(length)
+    val position = count
+    count += length
+    position
+  }
+
+  /**
+   * The array backing the buffered data. Only valid until the next write, reserve or hand-over, any of
+   * which may replace it.
+   */
+  def array: Array[Byte] = buf
+
+  /**
+   * Hands the buffered data over as a `ByteString` and leaves the stream empty for the next block.
+   *
+   * When most of the buffer is used the `ByteString` wraps the array without copying and the stream
+   * starts a fresh array for the next block; when only a small part is used the data is copied so the
+   * `ByteString` does not retain a large, mostly unused array, and the stream keeps its array.
+   */
+  def takeByteString(): ByteString =
     if (count < 1) ByteString.empty
-    else if (count > (buf.length >> 1))
-      // Most of the buffer is used — wrap it to avoid a copy
-      ByteString.fromArrayUnsafe(buf, 0, count)
-    else
-      // Small amount of data in a large buffer — copy to right-size so the rest can be GC'd
-      ByteString.fromArray(buf, 0, count)
+    else {
+      val result =
+        if (count > (buf.length >> 1)) {
+          val wrapped = ByteString.fromArrayUnsafe(buf, 0, count)
+          lastCount = count
+          buf = Array.emptyByteArray
+          wrapped
+        } else ByteString.fromArray(buf, 0, count)
+      count = 0
+      result
+    }
+
+  private def ensureCapacity(additional: Int): Unit = {
+    val required = count + additional
+    if (required > buf.length) {
+      val capacity =
+        if (buf.length == 0) math.max(math.max(initialCapacity, lastCount), required)
+        else math.max(buf.length << 1, required)
+      buf = Arrays.copyOf(buf, capacity)
+    }
+  }
 }
diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/util/StringTools.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/util/StringTools.scala
index 92823da04..3eb25e24b 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/impl/util/StringTools.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/impl/util/StringTools.scala
@@ -27,9 +27,9 @@ private[http] object StringTools {
     // keeps the array as is with a LATIN1 coder, so it is the same single copy.
     new String(bytes, ISO88591)
 
-  def asciiStringBytes(string: String): Array[Byte] = {
-    // this is as fast as Unsafe.copyUSAsciiStrToBytes for recent JDK versions
-    // and avoids the use of deprecated Unsafe methods
-    string.getBytes(java.nio.charset.StandardCharsets.US_ASCII)
-  }
+  def asciiStringBytes(string: String): Array[Byte] =
+    // ISO-8859-1 makes this the exact inverse of asciiStringFromBytes: HPACK string literals are opaque octets and
+    // a character in 0x80-0xFF has to come out as that octet, not as the '?' that encoding with US-ASCII would
+    // substitute. Since JDK 9 (compact strings) it is a single array copy for a string with a LATIN1 coder.
+    string.getBytes(ISO88591)
 }
diff --git a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/http2/hpack/HpackDecoderSpec.scala b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/http2/hpack/HpackDecoderSpec.scala
index 2e0d90711..bde40573b 100644
--- a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/http2/hpack/HpackDecoderSpec.scala
+++ b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/http2/hpack/HpackDecoderSpec.scala
@@ -17,11 +17,12 @@
 
 package org.apache.pekko.http.impl.engine.http2.hpack
 
-import java.io.{ ByteArrayInputStream, ByteArrayOutputStream, IOException, InputStream, SequenceInputStream }
+import java.io.{ ByteArrayInputStream, IOException, InputStream, SequenceInputStream }
 import scala.jdk.CollectionConverters._
 
 import scala.collection.mutable.ListBuffer
 
+import org.apache.pekko.http.impl.util.ByteStringOutputStream
 import org.apache.pekko.http.shaded.com.twitter.hpack.{ Decoder, Encoder, HeaderListener }
 
 import org.scalatest.matchers.should.Matchers
@@ -57,10 +58,10 @@ class HpackDecoderSpec extends AnyWordSpec with Matchers {
       bytes.grouped(chunkSize).map(chunk => new ByteArrayInputStream(chunk): InputStream).asJavaEnumeration)
 
   private def encode(headers: (String, String)*): Array[Byte] = {
-    val out = new ByteArrayOutputStream
+    val out = new ByteStringOutputStream(128)
     val encoder = new Encoder(maxHeaderTableSize)
     headers.foreach { case (name, value) => encoder.encodeHeader(out, name, value, false) }
-    out.toByteArray
+    out.takeByteString().toArray
   }
 
   private def decode(in: InputStream): Seq[(String, String)] = {
diff --git a/http-core/src/test/scala/org/apache/pekko/http/impl/util/ByteStringOutputStreamSpec.scala b/http-core/src/test/scala/org/apache/pekko/http/impl/util/ByteStringOutputStreamSpec.scala
index a9130162b..f8982e267 100644
--- a/http-core/src/test/scala/org/apache/pekko/http/impl/util/ByteStringOutputStreamSpec.scala
+++ b/http-core/src/test/scala/org/apache/pekko/http/impl/util/ByteStringOutputStreamSpec.scala
@@ -26,36 +26,53 @@ class ByteStringOutputStreamSpec extends AnyWordSpec with Matchers {
   "ByteStringOutputStream" must {
 
     "return an empty ByteString when nothing was written" in {
-      new ByteStringOutputStream(16).toByteStringUnsafe should ===(ByteString.empty)
+      new ByteStringOutputStream(16).takeByteString() should ===(ByteString.empty)
     }
 
     "return the bytes written when the buffer is exactly filled" in {
       val out = new ByteStringOutputStream(4)
       out.write(Array[Byte](1, 2, 3, 4))
-      out.toByteStringUnsafe should ===(ByteString(1, 2, 3, 4))
+      out.takeByteString() should ===(ByteString(1, 2, 3, 4))
     }
 
     "return the bytes written when the buffer was grown" in {
       val out = new ByteStringOutputStream(2)
       val data = Array.tabulate[Byte](1000)(i => i.toByte)
       out.write(data)
-      out.toByteStringUnsafe should ===(ByteString(data))
+      out.takeByteString() should ===(ByteString(data))
     }
 
     "return the bytes written when only a small part of the buffer is used" in {
       val out = new ByteStringOutputStream(1024)
       out.write(Array[Byte](1, 2, 3))
       out.write(4)
-      out.toByteStringUnsafe should ===(ByteString(1, 2, 3, 4))
+      out.takeByteString() should ===(ByteString(1, 2, 3, 4))
     }
 
     "not retain the buffer when only a small part of it is used" in {
       val out = new ByteStringOutputStream(1024)
       out.write(Array[Byte](1, 2, 3))
-      // the ByteString is a copy, so it is not affected by later writes to the stream
-      val result = out.toByteStringUnsafe
+      val array = out.array
+      // the ByteString is a copy, so it is not affected by later writes to the stream, which keeps its array
+      val result = out.takeByteString()
       out.write(Array[Byte](9, 9, 9))
       result should ===(ByteString(1, 2, 3))
+      (out.array eq array) shouldBe true
+    }
+
+    "hand over a mostly used buffer without copying and start the next block on a fresh array" in {
+      val out = new ByteStringOutputStream(4)
+      out.write(Array[Byte](1, 2, 3))
+      val array = out.array
+      val first = out.takeByteString()
+      first should ===(ByteString(1, 2, 3))
+      out.size shouldBe 0
+
+      // writes to the next block must not show up in the ByteString that was handed out
+      out.write(Array[Byte](9, 9, 9, 9), 0, 4)
+      (out.array eq array) shouldBe false
+      first should ===(ByteString(1, 2, 3))
+      out.takeByteString() should ===(ByteString(9, 9, 9, 9))
     }
 
     "write single bytes and byte ranges" in {
@@ -63,7 +80,25 @@ class ByteStringOutputStreamSpec extends AnyWordSpec with Matchers {
       out.write(1)
       out.write(Array[Byte](0, 2, 3, 0), 1, 2)
       out.write(Array[Byte](4, 5, 6, 7, 8))
-      out.toByteStringUnsafe should ===(ByteString(1, 2, 3, 4, 5, 6, 7, 8))
+      out.takeByteString() should ===(ByteString(1, 2, 3, 4, 5, 6, 7, 8))
+    }
+
+    "reserve a range that the caller fills in through the backing array" in {
+      val out = new ByteStringOutputStream(2)
+      out.write(7)
+      val position = out.reserve(3)
+      position shouldBe 1
+      out.size shouldBe 4
+      val array = out.array
+      array(position) = 1
+      array(position + 1) = 2
+      array(position + 2) = 3
+      out.write(8)
+      out.takeByteString() should ===(ByteString(7, 1, 2, 3, 8))
+    }
+
+    "reject a negative initial capacity" in {
+      an[IllegalArgumentException] should be thrownBy new ByteStringOutputStream(-1)
     }
   }
 }
diff --git a/http-core/src/test/scala/org/apache/pekko/http/impl/util/StringToolsSpec.scala b/http-core/src/test/scala/org/apache/pekko/http/impl/util/StringToolsSpec.scala
index ec30db525..2347668bf 100644
--- a/http-core/src/test/scala/org/apache/pekko/http/impl/util/StringToolsSpec.scala
+++ b/http-core/src/test/scala/org/apache/pekko/http/impl/util/StringToolsSpec.scala
@@ -50,5 +50,21 @@ class StringToolsSpec extends AnyWordSpec with Matchers {
     "encode an ASCII string to its US-ASCII bytes" in {
       StringTools.asciiStringBytes("abc") shouldEqual "abc".getBytes(StandardCharsets.US_ASCII)
     }
+
+    "be the inverse of asciiStringFromBytes for every octet" in {
+      // HPACK string literals are opaque octets, so a char in 0x80-0xFF must map back to that octet rather than
+      // to the '?' that encoding with US-ASCII would substitute
+      val allOctets = Array.tabulate(256)(_.toByte)
+      StringTools.asciiStringBytes(StringTools.asciiStringFromBytes(allOctets)) shouldEqual allOctets
+    }
+
+    "encode a character above 0xFF, which no octet can represent, as a single '?'" in {
+      // U+0100 forces the UTF-16 coder on JDK 9+, so this covers the non-arraycopy path too
+      StringTools.asciiStringBytes("aĀb") shouldEqual Array[Byte]('a', '?', 'b')
+    }
+
+    "encode the empty string to an empty array" in {
+      StringTools.asciiStringBytes("") shouldEqual Array.emptyByteArray
+    }
   }
 }
diff --git a/http-core/src/test/scala/org/apache/pekko/http/shaded/com/twitter/hpack/HpackEncoderSpec.scala b/http-core/src/test/scala/org/apache/pekko/http/shaded/com/twitter/hpack/HpackEncoderSpec.scala
new file mode 100644
index 000000000..8209603d0
--- /dev/null
+++ b/http-core/src/test/scala/org/apache/pekko/http/shaded/com/twitter/hpack/HpackEncoderSpec.scala
@@ -0,0 +1,128 @@
+/*
+ * 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.http.shaded.com.twitter.hpack
+
+import java.io.ByteArrayInputStream
+
+import scala.collection.immutable.VectorBuilder
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+import org.apache.pekko.http.impl.util.{ ByteStringOutputStream, StringTools }
+
+class HpackEncoderSpec extends AnyWordSpec with Matchers {
+
+  private def hex(bytes: Array[Byte]): String = bytes.map(b => f"$b%02x").mkString
+
+  private def huffmanEncode(bytes: Array[Byte]): Array[Byte] = {
+    val encodedLength = Huffman.ENCODER.getEncodedLength(bytes)
+    val dst = new Array[Byte](encodedLength)
+    Huffman.ENCODER.encode(bytes, dst, 0, encodedLength)
+    dst
+  }
+
+  private def huffmanEncode(literal: String): Array[Byte] = huffmanEncode(StringTools.asciiStringBytes(literal))
+
+  // indexing off so that the header value is always sent as a string literal, never as a table reference
+  private def encodeHeader(name: String, value: String, forceHuffmanOn: Boolean, forceHuffmanOff: Boolean)
+      : Array[Byte] = {
+    val encoder = new Encoder(4096, false, forceHuffmanOn, forceHuffmanOff)
+    val out = new ByteStringOutputStream(16)
+    encoder.encodeHeader(out, name, value, false)
+    out.takeByteString().toArray
+  }
+
+  private def decodeHeaders(bytes: Array[Byte]): Seq[(String, String)] = {
+    val decoder = new Decoder(8192, 4096)
+    val headers = new VectorBuilder[(String, String)]()
+    decoder.decode(new ByteArrayInputStream(bytes),
+      (name: String, value: String, parsedValue: AnyRef, _: Boolean) => {
+        headers += name -> value
+        parsedValue
+      })
+    headers.result()
+  }
+
+  "HuffmanEncoder" should {
+    // the string literals from RFC 7541 Appendix C.4 with their expected Huffman codes
+    val rfc7541Examples = Seq(
+      "www.example.com" -> "f1e3c2e5f23a6ba0ab90f4ff",
+      "no-cache" -> "a8eb10649cbf",
+      "custom-key" -> "25a849e95ba97d7f",
+      "custom-value" -> "25a849e95bb8e8b4bf")
+
+    "produce the Huffman codes from RFC 7541 Appendix C.4" in {
+      rfc7541Examples.foreach {
+        case (literal, expected) =>
+          hex(huffmanEncode(literal)) shouldEqual expected
+      }
+    }
+
+    "report the encoded length that encode actually produces" in {
+      rfc7541Examples.foreach {
+        case (literal, expected) =>
+          Huffman.ENCODER.getEncodedLength(StringTools.asciiStringBytes(literal)) shouldEqual expected.length / 2
+      }
+    }
+
+    "encode the empty literal to no bytes" in {
+      huffmanEncode("") shouldEqual Array.emptyByteArray
+      Huffman.ENCODER.getEncodedLength(Array.emptyByteArray) shouldEqual 0
+    }
+
+    "encode every octet with the code for that octet" in {
+      // a byte above 0x7F must be looked up as an unsigned symbol, not as a negative array index
+      val allOctets = Array.tabulate(256)(_.toByte)
+      Huffman.DECODER.decode(huffmanEncode(allOctets)) shouldEqual allOctets
+    }
+
+    "write at the given offset and reject an encoded length that does not match the input" in {
+      val bytes = StringTools.asciiStringBytes("no-cache")
+      val dst = new Array[Byte](10)
+      Huffman.ENCODER.encode(bytes, dst, 2, 6)
+      hex(dst) shouldEqual "0000a8eb10649cbf0000"
+      an[IllegalArgumentException] should be thrownBy Huffman.ENCODER.encode(bytes, new Array[Byte](10), 0, 7)
+      an[IndexOutOfBoundsException] should be thrownBy Huffman.ENCODER.encode(bytes, new Array[Byte](10), 5, 6)
+    }
+  }
+
+  "Encoder" should {
+    "encode the value from RFC 7541 Appendix C.4.1 with Huffman coding when it is shorter" in {
+      // 0x01: literal without indexing, name index 1 (:authority); 0x8c: Huffman coded, 12 octets
+      hex(encodeHeader(":authority", "www.example.com", forceHuffmanOn = false, forceHuffmanOff = false)) shouldEqual
+      "018cf1e3c2e5f23a6ba0ab90f4ff"
+    }
+
+    "encode the header from RFC 7541 Appendix C.2.1 as a raw literal when Huffman coding is forced off" in {
+      // 0x00: literal without indexing, new name; 0x0a/0x0d: raw string literals of 10 and 13 octets
+      hex(encodeHeader("custom-key", "custom-header", forceHuffmanOn = false, forceHuffmanOff = true)) shouldEqual
+      "000a637573746f6d2d6b65790d637573746f6d2d686561646572"
+    }
+
+    "round-trip a value carrying opaque octets above 0x7F in both string literal forms" in {
+      // HPACK string literals are opaque octets: a decoder maps each byte to the char of the same value, so an
+      // encoder has to map each such char back to that byte, in the raw form as well as the Huffman form
+      val value = new String(Array.tabulate(256)(_.toChar))
+      Seq(true, false).foreach { huffman =>
+        val bytes = encodeHeader("x-opaque", value, forceHuffmanOn = huffman, forceHuffmanOff = !huffman)
+        decodeHeaders(bytes) shouldEqual Seq("x-opaque" -> value)
+      }
+    }
+  }
+}
diff --git a/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/HPackEncodingSupport.scala b/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/HPackEncodingSupport.scala
index 063e868ff..4bd5912a6 100644
--- a/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/HPackEncodingSupport.scala
+++ b/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/HPackEncodingSupport.scala
@@ -13,12 +13,10 @@
 
 package org.apache.pekko.http.impl.engine.http2
 
-import java.io.ByteArrayOutputStream
-
 import org.apache.pekko
-import pekko.http.impl.util.StringRendering
 import pekko.http.scaladsl.model.{ HttpHeader, HttpRequest, HttpResponse }
 import pekko.http.scaladsl.model.headers.RawHeader
+import pekko.http.impl.util.{ ByteStringOutputStream, StringRendering }
 import pekko.http.shaded.com.twitter.hpack.Encoder
 import pekko.util.ByteString
 
@@ -62,12 +60,12 @@ trait HPackEncodingSupport {
     headers.map(h => h.lowercaseName -> h.value)
 
   def encodeHeaderPairs(headerPairs: Seq[(String, String)]): ByteString = {
-    val bos = new ByteArrayOutputStream()
+    val out = new ByteStringOutputStream(128)
 
-    def encode(name: String, value: String): Unit = encoder.encodeHeader(bos, name, value, false)
+    def encode(name: String, value: String): Unit = encoder.encodeHeader(out, name, value, false)
 
     headerPairs.foreach((encode _).tupled)
 
-    ByteString(bos.toByteArray)
+    out.takeByteString()
   }
 }