diff --git a/tornado/httputil.py b/tornado/httputil.py
index 4bd17786d..4d0d388f8 100644
--- a/tornado/httputil.py
+++ b/tornado/httputil.py
@@ -325,6 +325,14 @@ def parse(cls, headers: str, *, _chars_are_bytes: bool = True) -> HTTPHeaders:
     # MutableMapping abstract method implementations.
 
     def __setitem__(self, name: str, value: str) -> None:
+        # `MutableMapping` lets any object reach the indexer; non-string keys
+        # used to leak an `AttributeError` from inside `_normalize_header`
+        # (which is `lru_cache`d and splits on `-`). Reject them up-front with a
+        # clear message; `__contains__` already short-circuits the same way.
+        if not isinstance(name, str):
+            raise TypeError(
+                "HTTPHeaders keys must be str, not %s" % type(name).__name__
+            )
         norm_name = _normalize_header(name)
         self._combined_cache[norm_name] = value
         self._as_list[norm_name] = [value]
@@ -338,12 +346,20 @@ def __contains__(self, name: object) -> bool:
         return norm_name in self._as_list
 
     def __getitem__(self, name: str) -> str:
+        if not isinstance(name, str):
+            raise TypeError(
+                "HTTPHeaders keys must be str, not %s" % type(name).__name__
+            )
         header = _normalize_header(name)
         if header not in self._combined_cache:
             self._combined_cache[header] = ",".join(self._as_list[header])
         return self._combined_cache[header]
 
     def __delitem__(self, name: str) -> None:
+        if not isinstance(name, str):
+            raise TypeError(
+                "HTTPHeaders keys must be str, not %s" % type(name).__name__
+            )
         norm_name = _normalize_header(name)
         del self._combined_cache[norm_name]
         del self._as_list[norm_name]
diff --git a/tornado/test/httputil_test.py b/tornado/test/httputil_test.py
index 4e966eb50..d1bbce7d2 100644
--- a/tornado/test/httputil_test.py
+++ b/tornado/test/httputil_test.py
@@ -499,6 +499,48 @@ def test_setdefault(self):
         self.assertEqual(headers["quux"], "xyzzy")
         self.assertEqual(sorted(headers.get_all()), [("Foo", "bar"), ("Quux", "xyzzy")])
 
+    def test_non_string_key_setitem_raises_type_error(self):
+        # HTTPHeaders indexes via _normalize_header, which is decorated with
+        # @lru_cache and only accepts str. Non-string keys used to leak an
+        # AttributeError out of the cache wrapper; now they raise TypeError
+        # at the call site so callers get a useful error.
+        headers = HTTPHeaders()
+        for bad in (1, 1.5, None, b"Foo", ("Foo",), object()):
+            with self.assertRaises(TypeError):
+                headers[bad] = "value"
+            # __setitem__ must not silently store partial state for the bad
+            # key, and must not corrupt the lru_cache for valid lookups.
+            self.assertEqual(len(headers), 0)
+            self.assertNotIn(bad, headers)
+
+    def test_non_string_key_getitem_raises_type_error(self):
+        headers = HTTPHeaders()
+        headers["Foo"] = "bar"
+        for bad in (1, 1.5, None, b"Foo", ("Foo",), object()):
+            with self.assertRaises(TypeError):
+                headers[bad]
+        # Pre-existing string key still reads back.
+        self.assertEqual(headers["Foo"], "bar")
+
+    def test_non_string_key_delitem_raises_type_error(self):
+        headers = HTTPHeaders()
+        headers["Foo"] = "bar"
+        for bad in (1, 1.5, None, b"Foo", ("Foo",), object()):
+            with self.assertRaises(TypeError):
+                del headers[bad]
+        # Pre-existing string entry must still be intact.
+        self.assertEqual(headers["Foo"], "bar")
+
+    def test_non_string_key_contains_returns_false(self):
+        # __contains__ already guarded against non-strings; this pins the
+        # behaviour so a future refactor of __setitem__/__getitem__ cannot
+        # regress it.
+        headers = HTTPHeaders()
+        headers["Foo"] = "bar"
+        for bad in (1, 1.5, None, b"Foo", ("Foo",), object()):
+            self.assertFalse(bad in headers)
+        self.assertIn("Foo", headers)
+
     def test_string(self):
         headers = HTTPHeaders()
         headers.add("Foo", "1")
diff --git a/tornado/test/util_test.py b/tornado/test/util_test.py
index 83b78e5cf..45992e177 100644
--- a/tornado/test/util_test.py
+++ b/tornado/test/util_test.py
@@ -366,3 +366,42 @@ def test_version_info_compatible(self):
 
     def test_current_version(self):
         self.assert_version_info_compatible(tornado.version, tornado.version_info)
+
+
+class GzipDecompressorTest(unittest.TestCase):
+    def _compress(self, data: bytes) -> bytes:
+        import gzip
+
+        return gzip.compress(data)
+
+    def test_single_member(self) -> None:
+        data = b"hello world"
+        d = tornado.util.GzipDecompressor()
+        out = d.decompress(self._compress(data))
+        out += d.flush()
+        self.assertEqual(out, data)
+        self.assertEqual(d.unconsumed_tail, b"")
+
+    def test_concatenated_members_in_single_decompress(self) -> None:
+        members = [b"first message", b"second message", b"third message"]
+        blob = b"".join(self._compress(m) for m in members)
+        d = tornado.util.GzipDecompressor()
+        out = d.decompress(blob)
+        out += d.flush()
+        self.assertEqual(out, b"".join(members))
+
+    def test_concatenated_members_across_decompress_calls(self) -> None:
+        members = [b"alpha", b"beta", b"gamma", b"delta"]
+        blobs = [self._compress(m) for m in members]
+        d = tornado.util.GzipDecompressor()
+        out = bytearray()
+        # Split each member's bytes across separate decompress calls so we
+        # exercise the buffering path between calls.
+        for blob in blobs:
+            chunk_a = blob[: len(blob) // 2]
+            chunk_b = blob[len(blob) // 2 :]
+            out.extend(d.decompress(chunk_a))
+            out.extend(d.decompress(chunk_b))
+        out.extend(d.flush())
+        self.assertEqual(bytes(out), b"".join(members))
+        self.assertEqual(d.unconsumed_tail, b"")
diff --git a/tornado/util.py b/tornado/util.py
index 37b595e0b..6246dbf66 100644
--- a/tornado/util.py
+++ b/tornado/util.py
@@ -62,7 +62,13 @@ class GzipDecompressor:
     """Streaming gzip decompressor.
 
     The interface is like that of `zlib.decompressobj` (without some of the
-    optional arguments, but it understands gzip headers and checksums.
+    optional arguments, but it understands gzip headers and checksums).
+
+    .. versionchanged:: 6.6
+       Concatenated gzip members are now decompressed: after one member is
+       fully flushed, any remaining bytes in the stream are automatically
+       fed to a fresh decompressor for the next member. Previously, the
+       trailing members were silently dropped.
     """
 
     def __init__(self) -> None:
@@ -70,6 +76,10 @@ def __init__(self) -> None:
         # http://stackoverflow.com/questions/1838699/how-can-i-decompress-a-gzip-stream-with-zlib
         # This works on cpython and pypy, but not jython.
         self.decompressobj = zlib.decompressobj(16 + zlib.MAX_WBITS)
+        # Buffer for bytes that come in after the current member has been
+        # flushed; they belong to the next gzip member and are processed
+        # lazily on the next call to `decompress`.
+        self._pending: bytes = b""
 
     def decompress(self, value: bytes, max_length: int = 0) -> bytes:
         """Decompress a chunk, returning newly-available data.
@@ -81,21 +91,73 @@ def decompress(self, value: bytes, max_length: int = 0) -> bytes:
         If ``max_length`` is given, some input data may be left over
         in ``unconsumed_tail``; you must retrieve this value and pass
         it back to a future call to `decompress` if it is not empty.
+
+        .. versionchanged:: 6.6
+           If the previous member ended and ``value`` contains the start of
+           a new gzip member, the new member is decompressed transparently
+           rather than being buffered until the next call.
         """
-        return self.decompressobj.decompress(value, max_length)
+        # Combine any bytes left over from a previous call with the new
+        # input so we can advance through consecutive members in one pass.
+        if self._pending:
+            value = self._pending + value
+            self._pending = b""
+        result = bytearray()
+        if max_length == 0:
+            # Drain the entire stream, advancing across concatenated
+            # members transparently. This is the legacy behaviour that
+            # callers get when they don't impose a per-call cap.
+            while True:
+                chunk = self.decompressobj.decompress(value)
+                result.extend(chunk)
+                value = self.decompressobj.unconsumed_tail
+                if not self.decompressobj.eof:
+                    break
+                next_member = self.decompressobj.unused_data + value
+                if not next_member:
+                    break
+                self.decompressobj = zlib.decompressobj(16 + zlib.MAX_WBITS)
+                value = next_member
+            self._pending = b""
+            return bytes(result)
+        # When max_length is set the caller drives the loop via
+        # `unconsumed_tail`, so we return after a single inner call.
+        chunk = self.decompressobj.decompress(value, max_length)
+        result.extend(chunk)
+        trailing = self.decompressobj.unconsumed_tail
+        if self.decompressobj.eof:
+            # `unused_data` holds bytes past the current member's
+            # trailer that belong to the next concatenated member.
+            trailing = self.decompressobj.unused_data + trailing
+            self.decompressobj = zlib.decompressobj(16 + zlib.MAX_WBITS)
+        self._pending = trailing
+        return bytes(result)
 
     @property
     def unconsumed_tail(self) -> bytes:
         """Returns the unconsumed portion left over"""
-        return self.decompressobj.unconsumed_tail
+        return self._pending + self.decompressobj.unconsumed_tail
 
     def flush(self) -> bytes:
         """Return any remaining buffered data not yet returned by decompress.
 
         Also checks for errors such as truncated input.
         No other methods may be called on this object after `flush`.
+
+        .. versionchanged:: 6.6
+           If the stream contains concatenated gzip members, all members
+           are flushed and concatenated into the returned bytes.
         """
-        return self.decompressobj.flush()
+        result = bytearray()
+        while True:
+            result.extend(self.decompressobj.flush())
+            next_member = self.decompressobj.unused_data
+            self.decompressobj = zlib.decompressobj(16 + zlib.MAX_WBITS)
+            if not next_member:
+                break
+            # Drain the next member fully and append its output.
+            self.decompressobj.decompress(next_member)
+        return bytes(result)
 
 
 def import_object(name: str) -> Any: