Repository: GoogleCloudPlatform/pgadapter
Pull request #3180 (2025-04-16 15:16:58)

build(deps): bump com.google.cloud:google-cloud-shared-config from 1.15.1 to 1.15.4 in /samples/java/liquibase
Bumps [com.google.cloud:google-cloud-shared-config](https://github.com/googleapis/java-shared-config) from 1.15.1 to 1.15.4.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/googleapis/java-shared-config/releases">com.google.cloud:google-cloud-shared-config's releases</a>.</em></p>
<blockquote>
<h2>v1.15.4</h2>
<h2><a href="https://github.com/googleapis/java-shared-config/compare/v1.15.3...v1.15.4">1.15.4</a> (2025-04-15)</h2>
<h3>Bug Fixes</h3>
<ul>
<li>Use recommended variable expansion syntax (<a href="https://redirect.github.com/googleapis/java-shared-config/issues/1016">#1016</a>) (<a href="https://github.com/googleapis/java-shared-config/commit/6ff6dc48b8d92604164da7c68e268476b5a16578">6ff6dc4</a>)</li>
</ul>
<h2>v1.15.3</h2>
<h2><a href="https://github.com/googleapis/java-shared-config/compare/v1.15.2...v1.15.3">1.15.3</a> (2025-04-15)</h2>
<h3>Bug Fixes</h3>
<ul>
<li>Reintroduce support for GraalVM 17 (<a href="https://github.com/googleapis/java-shared-config/commit/75879e39eae3f119a40524d6ac8d22af8b8407d5">75879e3</a>)</li>
</ul>
<h2>v1.15.2</h2>
<h2><a href="https://github.com/googleapis/java-shared-config/compare/v1.15.1...v1.15.2">1.15.2</a> (2025-04-15)</h2>
<h3>Bug Fixes</h3>
<ul>
<li>Reintroduce support for GraalVM 17 (<a href="https://redirect.github.com/googleapis/java-shared-config/issues/1010">#1010</a>) (<a href="https://github.com/googleapis/java-shared-config/commit/19062614fd71a78c9cb88a7ff6f33b196ebba660">1906261</a>)</li>
</ul>
<h3>Dependencies</h3>
<ul>
<li>Update dependency org.graalvm.sdk:graal-sdk to v24.2.0 (<a href="https://redirect.github.com/googleapis/java-shared-config/issues/1005">#1005</a>) (<a href="https://github.com/googleapis/java-shared-config/commit/8186e31129eaa35257fd07cd23e8af4f1fc7742e">8186e31</a>)</li>
</ul>
</blockquote>
</details>
<details>
<summary>Changelog</summary>
<p><em>Sourced from <a href="https://github.com/googleapis/java-shared-config/blob/main/CHANGELOG.md">com.google.cloud:google-cloud-shared-config's changelog</a>.</em></p>
<blockquote>
<h2><a href="https://github.com/googleapis/java-shared-config/compare/v1.15.3...v1.15.4">1.15.4</a> (2025-04-15)</h2>
<h3>Bug Fixes</h3>
<ul>
<li>Use recommended variable expansion syntax (<a href="https://redirect.github.com/googleapis/java-shared-config/issues/1016">#1016</a>) (<a href="https://github.com/googleapis/java-shared-config/commit/6ff6dc48b8d92604164da7c68e268476b5a16578">6ff6dc4</a>)</li>
</ul>
<h2><a href="https://github.com/googleapis/java-shared-config/compare/v1.15.2...v1.15.3">1.15.3</a> (2025-04-15)</h2>
<h3>Bug Fixes</h3>
<ul>
<li>Reintroduce support for GraalVM 17 (<a href="https://github.com/googleapis/java-shared-config/commit/75879e39eae3f119a40524d6ac8d22af8b8407d5">75879e3</a>)</li>
</ul>
<h2><a href="https://github.com/googleapis/java-shared-config/compare/v1.15.1...v1.15.2">1.15.2</a> (2025-04-15)</h2>
<h3>Bug Fixes</h3>
<ul>
<li>Reintroduce support for GraalVM 17 (<a href="https://redirect.github.com/googleapis/java-shared-config/issues/1010">#1010</a>) (<a href="https://github.com/googleapis/java-shared-config/commit/19062614fd71a78c9cb88a7ff6f33b196ebba660">1906261</a>)</li>
</ul>
<h3>Dependencies</h3>
<ul>
<li>Update dependency org.graalvm.sdk:graal-sdk to v24.2.0 (<a href="https://redirect.github.com/googleapis/java-shared-config/issues/1005">#1005</a>) (<a href="https://github.com/googleapis/java-shared-config/commit/8186e31129eaa35257fd07cd23e8af4f1fc7742e">8186e31</a>)</li>
</ul>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/googleapis/java-shared-config/commit/94f780d073b99c66e4ed81d014d0ac1763e85e55"><code>94f780d</code></a> chore(main): release 1.15.4 (<a href="https://redirect.github.com/googleapis/java-shared-config/issues/1017">#1017</a>)</li>
<li><a href="https://github.com/googleapis/java-shared-config/commit/6ff6dc48b8d92604164da7c68e268476b5a16578"><code>6ff6dc4</code></a> fix: use recommended variable expansion syntax (<a href="https://redirect.github.com/googleapis/java-shared-config/issues/1016">#1016</a>)</li>
<li><a href="https://github.com/googleapis/java-shared-config/commit/822c2376ec31108e7ab5a02c07dbd31ef1b897fc"><code>822c237</code></a> chore(main): release 1.15.4-SNAPSHOT (<a href="https://redirect.github.com/googleapis/java-shared-config/issues/1015">#1015</a>)</li>
<li><a href="https://github.com/googleapis/java-shared-config/commit/1d3d74f741799a43bda30e3155ec59c2a80b6889"><code>1d3d74f</code></a> chore(main): release 1.15.3 (<a href="https://redirect.github.com/googleapis/java-shared-config/issues/1014">#1014</a>)</li>
<li><a href="https://github.com/googleapis/java-shared-config/commit/75879e39eae3f119a40524d6ac8d22af8b8407d5"><code>75879e3</code></a> chore(main): release 1.15.3-SNAPSHOT (<a href="https://redirect.github.com/googleapis/java-shared-config/issues/1012">#1012</a>)</li>
<li><a href="https://github.com/googleapis/java-shared-config/commit/74c88ae12c0ce5558468f4c8a5b5948cd45a4361"><code>74c88ae</code></a> chore(main): release 1.15.2 (<a href="https://redirect.github.com/googleapis/java-shared-config/issues/1011">#1011</a>)</li>
<li><a href="https://github.com/googleapis/java-shared-config/commit/55ab15f65a161253a53f1185808c980a9833a438"><code>55ab15f</code></a> chore(main): release 1.15.2-SNAPSHOT (<a href="https://redirect.github.com/googleapis/java-shared-config/issues/1007">#1007</a>)</li>
<li><a href="https://github.com/googleapis/java-shared-config/commit/19062614fd71a78c9cb88a7ff6f33b196ebba660"><code>1906261</code></a> fix: reintroduce support for GraalVM 17 (<a href="https://redirect.github.com/googleapis/java-shared-config/issues/1010">#1010</a>)</li>
<li><a href="https://github.com/googleapis/java-shared-config/commit/8186e31129eaa35257fd07cd23e8af4f1fc7742e"><code>8186e31</code></a> deps: update dependency org.graalvm.sdk:graal-sdk to v24.2.0 (<a href="https://redirect.github.com/googleapis/java-shared-config/issues/1005">#1005</a>)</li>
<li>See full diff in <a href="https://github.com/googleapis/java-shared-config/compare/v1.15.1...v1.15.4">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=com.google.cloud:google-cloud-shared-config&package-manager=maven&previous-version=1.15.1&new-version=1.15.4)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot show <dependency name> ignore conditions` will show all of the ignore conditions of the specified dependency
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>


Patch:
diff --git a/benchmarks/latency-comparison/golang/go.mod b/benchmarks/latency-comparison/golang/go.mod
index a8a90bf0..c8fa8048 100644
--- a/benchmarks/latency-comparison/golang/go.mod
+++ b/benchmarks/latency-comparison/golang/go.mod
@@ -6,7 +6,7 @@ toolchain go1.24.2
 
 require (
 	cloud.google.com/go/spanner v1.79.0
-	github.com/docker/docker v28.0.4+incompatible
+	github.com/docker/docker v28.1.1+incompatible
 	github.com/jackc/pgx/v4 v4.18.3
 	github.com/jackc/pgx/v5 v5.7.4
 	github.com/montanaflynn/stats v0.7.0
@@ -23,7 +23,7 @@ require (
 	cloud.google.com/go/compute/metadata v0.6.0 // indirect
 	cloud.google.com/go/monitoring v1.24.1 // indirect
 	dario.cat/mergo v1.0.1 // indirect
-	github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24 // indirect
+	github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 // indirect
 	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
 	github.com/GoogleCloudPlatform/grpc-gcp-go/grpcgcp v1.5.2 // indirect
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.27.0 // indirect
@@ -58,13 +58,14 @@ require (
 	github.com/jackc/pgproto3/v2 v2.3.3 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
 	github.com/jackc/pgtype v1.14.0 // indirect
-	github.com/klauspost/compress v1.17.4 // indirect
+	github.com/klauspost/compress v1.18.0 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
 	github.com/magiconair/properties v1.8.9 // indirect
 	github.com/moby/docker-image-spec v1.3.1 // indirect
+	github.com/moby/go-archive v0.1.0 // indirect
 	github.com/moby/patternmatcher v0.6.0 // indirect
-	github.com/moby/sys/sequential v0.5.0 // indirect
-	github.com/moby/sys/user v0.3.0 // indirect
+	github.com/moby/sys/sequential v0.6.0 // indirect
+	github.com/moby/sys/user v0.4.0 // indirect
 	github.com/moby/sys/userns v0.1.0 // indirect
 	github.com/moby/term v0.5.0 // indirect
 	github.com/morikuni/aec v1.0.0 // indirect
diff --git a/benchmarks/latency-comparison/golang/go.sum b/benchmarks/latency-comparison/golang/go.sum
index 63f69e87..e49c537b 100644
--- a/benchmarks/latency-comparison/golang/go.sum
+++ b/benchmarks/latency-comparison/golang/go.sum
@@ -633,6 +633,8 @@ gioui.org v0.0.0-20210308172011-57750fc8a0a6/go.mod h1:RSH6KIUZ0p2xy5zHDxgAM4zum
 git.sr.ht/~sbinet/gg v0.3.1/go.mod h1:KGYtlADtqsqANL9ueOFkWymvzUvLMQllU5Ixo+8v3pc=
 github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24 h1:bvDV9vkmnHYOMsOr4WLk+Vo07yKIzd94sVoIqshQ4bU=
 github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8=
+github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 h1:He8afgbRMd7mFxO99hRNu+6tazq8nFF9lIwo9JFroBk=
+github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8=
 github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 h1:UQHMgLO+TxOElx5B5HZ4hJQsoJ/PvUvKRhJHDQXO8P8=
 github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
@@ -715,6 +717,8 @@ github.com/docker/docker v28.0.2+incompatible h1:9BILleFwug5FSSqWBgVevgL3ewDJfWW
 github.com/docker/docker v28.0.2+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/docker v28.0.4+incompatible h1:JNNkBctYKurkw6FrHfKqY0nKIDf5nrbxjVBtS+cdcok=
 github.com/docker/docker v28.0.4+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
+github.com/docker/docker v28.1.1+incompatible h1:49M11BFLsVO1gxY9UX9p/zwkE/rswggs8AdFmXQw51I=
+github.com/docker/docker v28.1.1+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/go-connections v0.5.0 h1:USnMq7hx7gwdVZq1L49hLXaFtUdTADjXGp+uj1Br63c=
 github.com/docker/go-connections v0.5.0/go.mod h1:ov60Kzw0kKElRwhNs9UlUHAE/F9Fe6GLaXnqyDdmEXc=
 github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
@@ -970,6 +974,8 @@ github.com/klauspost/asmfmt v1.3.2/go.mod h1:AG8TuvYojzulgDAMCnYn50l/5QV3Bs/tp6j
 github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
 github.com/klauspost/compress v1.17.4 h1:Ej5ixsIri7BrIjBkRZLTo6ghwrEtHFk7ijlczPW4fZ4=
 github.com/klauspost/compress v1.17.4/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM=
+github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
+github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
@@ -1009,12 +1015,18 @@ github.com/minio/asm2plan9s v0.0.0-20200509001527-cdd76441f9d8/go.mod h1:mC1jAcs
 github.com/minio/c2goasm v0.0.0-20190812172519-36a3d3bbc4f3/go.mod h1:RagcQ7I8IeTMnF8JTXieKnO4Z6JCsikNEzj0DwauVzE=
 github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0=
 github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo=
+github.com/moby/go-archive v0.1.0 h1:Kk/5rdW/g+H8NHdJW2gsXyZ7UnzvJNOy6VKJqueWdcQ=
+github.com/moby/go-archive v0.1.0/go.mod h1:G9B+YoujNohJmrIYFBpSd54GTUB4lt9S+xVQvsJyFuo=
 github.com/moby/patternmatcher v0.6.0 h1:GmP9lR19aU5GqSSFko+5pRqHi+Ohk1O69aFiKkVGiPk=
 github.com/moby/patternmatcher v0.6.0/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YOUYxkhApJIxc=
 github.com/moby/sys/sequential v0.5.0 h1:OPvI35Lzn9K04PBbCLW0g4LcFAJgHsvXsRyewg5lXtc=
 github.com/moby/sys/sequential v0.5.0/go.mod h1:tH2cOOs5V9MlPiXcQzRC+eEyab644PWKGRYaaV5ZZlo=
+github.com/moby/sys/sequential v0.6.0 h1:qrx7XFUd/5DxtqcoH1h438hF5TmOvzC/lspjy7zgvCU=
+github.com/moby/sys/sequential v0.6.0/go.mod h1:uyv8EUTrca5PnDsdMGXhZe6CCe8U/UiTWd+lL+7b/Ko=
 github.com/moby/sys/user v0.3.0 h1:9ni5DlcW5an3SvRSx4MouotOygvzaXbaSrc/wGDFWPo=
 github.com/moby/sys/user v0.3.0/go.mod h1:bG+tYYYJgaMtRKgEmuueC0hJEAZWwtIbZTB+85uoHjs=
+github.com/moby/sys/user v0.4.0 h1:jhcMKit7SA80hivmFJcbB1vqmw//wU61Zdui2eQXuMs=
+github.com/moby/sys/user v0.4.0/go.mod h1:bG+tYYYJgaMtRKgEmuueC0hJEAZWwtIbZTB+85uoHjs=
 github.com/moby/sys/userns v0.1.0 h1:tVLXkFOxVu9A64/yh59slHVv9ahO9UIev4JZusOLG/g=
 github.com/moby/sys/userns v0.1.0/go.mod h1:IHUYgu/kao6N8YZlp9Cf444ySSvCmDlmzUcYfDHOl28=
 github.com/moby/term v0.5.0 h1:xt8Q1nalod/v7BqbG21f8mQPqH+xAaC9C3N3wfWbVP0=
@@ -1890,6 +1902,7 @@ gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gotest.tools/v3 v3.5.1 h1:EENdUnS3pdur5nybKYIh2Vfgc8IUNBjxDPSjtiJcOzU=
 gotest.tools/v3 v3.5.1/go.mod h1:isy3WKz7GK6uNw/sbHzfKBLvlvXwUyV06n6brMxxopU=
+gotest.tools/v3 v3.5.2 h1:7koQfIKdy+I8UTetycgUqXWSDwpgv193Ka+qRsmBY8Q=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190106161140-3f1c8253044a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190418001031-e561f6794a2a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
diff --git a/benchmarks/latency-comparison/java/pom.xml b/benchmarks/latency-comparison/java/pom.xml
index 42fd088f..5cb3961b 100644
--- a/benchmarks/latency-comparison/java/pom.xml
+++ b/benchmarks/latency-comparison/java/pom.xml
@@ -35,7 +35,7 @@
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-shared-config</artifactId>
-    <version>1.15.1</version>
+    <version>1.15.4</version>
   </parent>
 
   <dependencyManagement>
diff --git a/pom.xml b/pom.xml
index acaa2834..772471af 100644
--- a/pom.xml
+++ b/pom.xml
@@ -55,7 +55,7 @@
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-shared-config</artifactId>
-    <version>1.15.1</version>
+    <version>1.15.4</version>
   </parent>
   <scm>
     <connection>scm:git:git@github.com:GoogleCloudPlatform/pgadapter.git</connection>
diff --git a/samples/java/liquibase/pom.xml b/samples/java/liquibase/pom.xml
index 8981b91f..11904c29 100644
--- a/samples/java/liquibase/pom.xml
+++ b/samples/java/liquibase/pom.xml
@@ -37,7 +37,7 @@
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-shared-config</artifactId>
-    <version>1.15.1</version>
+    <version>1.15.4</version>
   </parent>
   <scm>
     <connection>scm:git:git@github.com:GoogleCloudPlatform/pgadapter.git</connection>
diff --git a/samples/snippets/golang-snippets/go.mod b/samples/snippets/golang-snippets/go.mod
index bffa7d7e..4987461b 100644
--- a/samples/snippets/golang-snippets/go.mod
+++ b/samples/snippets/golang-snippets/go.mod
@@ -5,14 +5,14 @@ go 1.23.0
 toolchain go1.24.2
 
 require (
-	github.com/docker/docker v28.0.4+incompatible
+	github.com/docker/docker v28.1.1+incompatible
 	github.com/jackc/pgx/v5 v5.7.4
 	github.com/testcontainers/testcontainers-go v0.36.0
 )
 
 require (
 	dario.cat/mergo v1.0.1 // indirect
-	github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24 // indirect
+	github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 // indirect
 	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
 	github.com/Microsoft/go-winio v0.6.2 // indirect
 	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
@@ -32,13 +32,14 @@ require (
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
-	github.com/klauspost/compress v1.17.4 // indirect
+	github.com/klauspost/compress v1.18.0 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
 	github.com/magiconair/properties v1.8.9 // indirect
 	github.com/moby/docker-image-spec v1.3.1 // indirect
+	github.com/moby/go-archive v0.1.0 // indirect
 	github.com/moby/patternmatcher v0.6.0 // indirect
-	github.com/moby/sys/sequential v0.5.0 // indirect
-	github.com/moby/sys/user v0.3.0 // indirect
+	github.com/moby/sys/sequential v0.6.0 // indirect
+	github.com/moby/sys/user v0.4.0 // indirect
 	github.com/moby/sys/userns v0.1.0 // indirect
 	github.com/moby/term v0.5.0 // indirect
 	github.com/morikuni/aec v1.0.0 // indirect
diff --git a/samples/snippets/golang-snippets/go.sum b/samples/snippets/golang-snippets/go.sum
index 8fa46593..00669faa 100644
--- a/samples/snippets/golang-snippets/go.sum
+++ b/samples/snippets/golang-snippets/go.sum
@@ -4,6 +4,7 @@ dario.cat/mergo v1.0.1 h1:Ra4+bf83h2ztPIQYNP99R6m+Y7KfnARDfID+a+vLl4s=
 dario.cat/mergo v1.0.1/go.mod h1:uNxQE+84aUszobStD9th8a29P2fMDhsBdgRYvZOxGmk=
 github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24 h1:bvDV9vkmnHYOMsOr4WLk+Vo07yKIzd94sVoIqshQ4bU=
 github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8=
+github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8=
 github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 h1:UQHMgLO+TxOElx5B5HZ4hJQsoJ/PvUvKRhJHDQXO8P8=
 github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E=
 github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY=
@@ -27,6 +28,8 @@ github.com/docker/docker v28.0.2+incompatible h1:9BILleFwug5FSSqWBgVevgL3ewDJfWW
 github.com/docker/docker v28.0.2+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/docker v28.0.4+incompatible h1:JNNkBctYKurkw6FrHfKqY0nKIDf5nrbxjVBtS+cdcok=
 github.com/docker/docker v28.0.4+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
+github.com/docker/docker v28.1.1+incompatible h1:49M11BFLsVO1gxY9UX9p/zwkE/rswggs8AdFmXQw51I=
+github.com/docker/docker v28.1.1+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/go-connections v0.5.0 h1:USnMq7hx7gwdVZq1L49hLXaFtUdTADjXGp+uj1Br63c=
 github.com/docker/go-connections v0.5.0/go.mod h1:ov60Kzw0kKElRwhNs9UlUHAE/F9Fe6GLaXnqyDdmEXc=
 github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
@@ -67,6 +70,8 @@ github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.17.4 h1:Ej5ixsIri7BrIjBkRZLTo6ghwrEtHFk7ijlczPW4fZ4=
 github.com/klauspost/compress v1.17.4/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM=
+github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
+github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
@@ -79,12 +84,18 @@ github.com/magiconair/properties v1.8.9 h1:nWcCbLq1N2v/cpNsy5WvQ37Fb+YElfq20WJ/a
 github.com/magiconair/properties v1.8.9/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0=
 github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0=
 github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo=
+github.com/moby/go-archive v0.1.0 h1:Kk/5rdW/g+H8NHdJW2gsXyZ7UnzvJNOy6VKJqueWdcQ=
+github.com/moby/go-archive v0.1.0/go.mod h1:G9B+YoujNohJmrIYFBpSd54GTUB4lt9S+xVQvsJyFuo=
 github.com/moby/patternmatcher v0.6.0 h1:GmP9lR19aU5GqSSFko+5pRqHi+Ohk1O69aFiKkVGiPk=
 github.com/moby/patternmatcher v0.6.0/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YOUYxkhApJIxc=
 github.com/moby/sys/sequential v0.5.0 h1:OPvI35Lzn9K04PBbCLW0g4LcFAJgHsvXsRyewg5lXtc=
 github.com/moby/sys/sequential v0.5.0/go.mod h1:tH2cOOs5V9MlPiXcQzRC+eEyab644PWKGRYaaV5ZZlo=
+github.com/moby/sys/sequential v0.6.0 h1:qrx7XFUd/5DxtqcoH1h438hF5TmOvzC/lspjy7zgvCU=
+github.com/moby/sys/sequential v0.6.0/go.mod h1:uyv8EUTrca5PnDsdMGXhZe6CCe8U/UiTWd+lL+7b/Ko=
 github.com/moby/sys/user v0.3.0 h1:9ni5DlcW5an3SvRSx4MouotOygvzaXbaSrc/wGDFWPo=
 github.com/moby/sys/user v0.3.0/go.mod h1:bG+tYYYJgaMtRKgEmuueC0hJEAZWwtIbZTB+85uoHjs=
+github.com/moby/sys/user v0.4.0 h1:jhcMKit7SA80hivmFJcbB1vqmw//wU61Zdui2eQXuMs=
+github.com/moby/sys/user v0.4.0/go.mod h1:bG+tYYYJgaMtRKgEmuueC0hJEAZWwtIbZTB+85uoHjs=
 github.com/moby/sys/userns v0.1.0 h1:tVLXkFOxVu9A64/yh59slHVv9ahO9UIev4JZusOLG/g=
 github.com/moby/sys/userns v0.1.0/go.mod h1:IHUYgu/kao6N8YZlp9Cf444ySSvCmDlmzUcYfDHOl28=
 github.com/moby/term v0.5.0 h1:xt8Q1nalod/v7BqbG21f8mQPqH+xAaC9C3N3wfWbVP0=
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java b/src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java
index 5fd96fd7..e895711a 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java
@@ -26,6 +26,7 @@ import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata.TextFormat;
 import com.google.cloud.spanner.pgadapter.statements.IntermediateStatement;
 import com.google.cloud.spanner.pgadapter.utils.Metrics;
+import com.google.cloud.spanner.pgadapter.wireprotocol.MessageReader;
 import com.google.cloud.spanner.pgadapter.wireprotocol.WireMessage;
 import com.google.common.collect.ImmutableList;
 import io.opentelemetry.api.OpenTelemetry;
@@ -69,6 +70,7 @@ public class ProxyServer extends AbstractApiService {
   private final Metrics metrics;
   private final Properties properties;
   private final List<ConnectionHandler> handlers = new LinkedList<>();
+  private final MessageReader messageReader;
 
   /**
    * Latch that is closed when the TCP server has started. We need this to know the exact port that
@@ -162,6 +164,7 @@ public class ProxyServer extends AbstractApiService {
   public ProxyServer(
       OptionsMetadata optionsMetadata, OpenTelemetry openTelemetry, Properties properties) {
     this.options = optionsMetadata;
+    this.messageReader = new MessageReader(optionsMetadata);
     this.openTelemetry = openTelemetry;
     this.metrics =
         optionsMetadata.isEnableOpenTelemetryMetrics()
@@ -176,6 +179,10 @@ public class ProxyServer extends AbstractApiService {
     addConnectionProperties();
   }
 
+  public MessageReader getMessageReader() {
+    return this.messageReader;
+  }
+
   private void addConnectionProperties() {
     for (Map.Entry<String, String> entry : options.getPropertyMap().entrySet()) {
       properties.setProperty(entry.getKey(), entry.getValue());
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/BindMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/BindMessage.java
index a1291671..35d28d88 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/BindMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/BindMessage.java
@@ -47,9 +47,9 @@ public class BindMessage extends AbstractQueryProtocolMessage {
     super(connection);
     this.portalName = this.readString();
     this.statementName = this.readString();
-    this.formatCodes = getFormatCodes(this.inputStream);
+    this.formatCodes = MessageReader.getFormatCodes(this.inputStream);
     this.parameters = getParameters(this.inputStream);
-    this.resultFormatCodes = getFormatCodes(this.inputStream);
+    this.resultFormatCodes = MessageReader.getFormatCodes(this.inputStream);
     IntermediatePreparedStatement statement = connection.getStatement(statementName);
     this.statement =
         statement.createPortal(
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ControlMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ControlMessage.java
index bd4dff09..6ea74df6 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ControlMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ControlMessage.java
@@ -15,7 +15,6 @@
 package com.google.cloud.spanner.pgadapter.wireprotocol;
 
 import static com.google.cloud.spanner.pgadapter.statements.BackendConnection.DB_STATEMENT;
-import static com.google.cloud.spanner.pgadapter.statements.IntermediatePortalStatement.NO_FORMAT_CODES;
 
 import com.google.api.core.InternalApi;
 import com.google.api.gax.grpc.GrpcCallContext;
@@ -32,12 +31,8 @@ import com.google.cloud.spanner.connection.Connection;
 import com.google.cloud.spanner.connection.StatementResult;
 import com.google.cloud.spanner.connection.StatementResult.ResultType;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
-import com.google.cloud.spanner.pgadapter.ConnectionHandler.ConnectionStatus;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.QueryMode;
-import com.google.cloud.spanner.pgadapter.error.PGException;
 import com.google.cloud.spanner.pgadapter.error.PGExceptionFactory;
-import com.google.cloud.spanner.pgadapter.error.SQLState;
-import com.google.cloud.spanner.pgadapter.error.Severity;
 import com.google.cloud.spanner.pgadapter.metadata.SendResultSetState;
 import com.google.cloud.spanner.pgadapter.statements.BackendConnection.PartitionQueryResult;
 import com.google.cloud.spanner.pgadapter.statements.CopyToStatement;
@@ -61,7 +56,6 @@ import io.grpc.MethodDescriptor;
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.api.trace.Tracer;
 import io.opentelemetry.context.Scope;
-import java.io.DataInputStream;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
@@ -109,117 +103,6 @@ public abstract class ControlMessage extends WireMessage {
     return manuallyCreatedToken == null;
   }
 
-  /**
-   * Factory method to create the message from the specific command type char.
-   *
-   * @param connection The connection handler object setup with the ability to send/receive.
-   * @return The constructed wire message given the input message.
-   * @throws Exception If construction or reading fails.
-   */
-  public static ControlMessage create(ConnectionHandler connection) throws Exception {
-    boolean validMessage = true;
-    char nextMsg = (char) connection.getConnectionMetadata().getInputStream().readUnsignedByte();
-    try {
-      if (connection.getStatus() == ConnectionStatus.COPY_IN) {
-        switch (nextMsg) {
-          case CopyDoneMessage.IDENTIFIER:
-            return new CopyDoneMessage(connection);
-          case CopyDataMessage.IDENTIFIER:
-            return new CopyDataMessage(connection);
-          case CopyFailMessage.IDENTIFIER:
-            return new CopyFailMessage(connection);
-          case SyncMessage.IDENTIFIER:
-          case FlushMessage.IDENTIFIER:
-            // Skip sync/flush in COPY_IN. This is consistent with real PostgreSQL which also does
-            // this to accommodate clients that do not check what type of statement they sent in an
-            // ExecuteMessage, and instead always blindly send a flush/sync after each execute.
-            return SkipMessage.createForValidStream(connection);
-          default:
-            // Skip other unexpected messages and throw an exception to fail the copy operation.
-            validMessage = false;
-            SkipMessage.createForInvalidStream(connection);
-            throw new IllegalStateException(
-                String.format(
-                    "Expected CopyData ('d'), CopyDone ('c') or CopyFail ('f') messages, got: '%c'",
-                    nextMsg));
-        }
-      } else {
-        switch (nextMsg) {
-          case QueryMessage.IDENTIFIER:
-            return new QueryMessage(connection);
-          case ParseMessage.IDENTIFIER:
-            return new ParseMessage(connection);
-          case BindMessage.IDENTIFIER:
-            return new BindMessage(connection);
-          case DescribeMessage.IDENTIFIER:
-            return new DescribeMessage(connection);
-          case ExecuteMessage.IDENTIFIER:
-            return new ExecuteMessage(connection);
-          case CloseMessage.IDENTIFIER:
-            return new CloseMessage(connection);
-          case TerminateMessage.IDENTIFIER:
-            return new TerminateMessage(connection);
-          case FunctionCallMessage.IDENTIFIER:
-            return new FunctionCallMessage(connection);
-          case FlushMessage.IDENTIFIER:
-            return new FlushMessage(connection);
-          case SyncMessage.IDENTIFIER:
-            return new SyncMessage(connection);
-          case CopyDoneMessage.IDENTIFIER:
-          case CopyDataMessage.IDENTIFIER:
-          case CopyFailMessage.IDENTIFIER:
-            // Silently skip COPY messages in non-COPY mode. This is consistent with the PG wire
-            // protocol. If we continue to receive COPY messages while in non-COPY mode, we'll
-            // terminate the connection to prevent the server from being flooded with invalid
-            // messages.
-            validMessage = false;
-            // Note: The stream itself is still valid as we received a message that we recognized.
-            return SkipMessage.createForValidStream(connection);
-          default:
-            throw new IllegalStateException(String.format("Unknown message: %c", nextMsg));
-        }
-      }
-    } finally {
-      if (validMessage) {
-        connection.clearInvalidMessageCount();
-      } else {
-        connection.increaseInvalidMessageCount();
-        if (connection.getInvalidMessageCount() > MAX_INVALID_MESSAGE_COUNT) {
-          new ErrorResponse(
-                  connection,
-                  PGException.newBuilder(
-                          String.format(
-                              "Received %d invalid/unexpected messages. Last received message: '%c'",
-                              connection.getInvalidMessageCount(), nextMsg))
-                      .setSQLState(SQLState.ProtocolViolation)
-                      .setSeverity(Severity.FATAL)
-                      .build())
-              .send();
-          connection.setStatus(ConnectionStatus.TERMINATED);
-        }
-      }
-    }
-  }
-
-  /**
-   * Extract format codes from message (useful for both input and output format codes).
-   *
-   * @param input The data stream containing the user request.
-   * @return A list of format codes.
-   * @throws Exception If reading fails in any way.
-   */
-  protected static short[] getFormatCodes(DataInputStream input) throws Exception {
-    short numberOfFormatCodes = input.readShort();
-    if (numberOfFormatCodes == 0) {
-      return NO_FORMAT_CODES;
-    }
-    short[] formatCodes = new short[numberOfFormatCodes];
-    for (int i = 0; i < numberOfFormatCodes; i++) {
-      formatCodes[i] = input.readShort();
-    }
-    return formatCodes;
-  }
-
   public enum PreparedType {
     Portal,
     Statement;
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/FunctionCallMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/FunctionCallMessage.java
index bbdee289..49af2714 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/FunctionCallMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/FunctionCallMessage.java
@@ -37,7 +37,7 @@ public class FunctionCallMessage extends ControlMessage {
   public FunctionCallMessage(ConnectionHandler connection) throws Exception {
     super(connection);
     this.functionID = this.inputStream.readInt();
-    this.argumentFormatCodes = getFormatCodes(this.inputStream);
+    this.argumentFormatCodes = MessageReader.getFormatCodes(this.inputStream);
     this.arguments = getParameters(this.inputStream);
     this.resultFormatCode = this.inputStream.readShort();
   }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/MessageReader.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/MessageReader.java
new file mode 100644
index 00000000..a173ca14
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/MessageReader.java
@@ -0,0 +1,158 @@
+// Copyright 2025 Google LLC
+//
+// Licensed 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 com.google.cloud.spanner.pgadapter.wireprotocol;
+
+import static com.google.cloud.spanner.pgadapter.statements.IntermediatePortalStatement.NO_FORMAT_CODES;
+
+import com.google.api.core.InternalApi;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler.ConnectionStatus;
+import com.google.cloud.spanner.pgadapter.error.PGException;
+import com.google.cloud.spanner.pgadapter.error.SQLState;
+import com.google.cloud.spanner.pgadapter.error.Severity;
+import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
+import com.google.cloud.spanner.pgadapter.wireoutput.ErrorResponse;
+import java.io.DataInputStream;
+import java.io.IOException;
+
+/** This class reads the next control message for a connection. */
+@InternalApi
+public class MessageReader {
+  /** Maximum number of invalid messages in a row allowed before we terminate the connection. */
+  static final int MAX_INVALID_MESSAGE_COUNT = 50;
+
+  private final OptionsMetadata options;
+
+  public MessageReader(OptionsMetadata options) {
+    this.options = options;
+  }
+
+  /**
+   * Factory method to create the next control message for the given connection.
+   *
+   * @param connection The connection handler object setup with the ability to send/receive.
+   * @return The constructed wire message given the input message.
+   * @throws Exception If construction or reading fails.
+   */
+  public ControlMessage create(ConnectionHandler connection) throws Exception {
+    boolean validMessage = true;
+    char nextMsg = readNextMsgIdentifier(connection);
+    try {
+      if (connection.getStatus() == ConnectionStatus.COPY_IN) {
+        switch (nextMsg) {
+          case CopyDoneMessage.IDENTIFIER:
+            return new CopyDoneMessage(connection);
+          case CopyDataMessage.IDENTIFIER:
+            return new CopyDataMessage(connection);
+          case CopyFailMessage.IDENTIFIER:
+            return new CopyFailMessage(connection);
+          case SyncMessage.IDENTIFIER:
+          case FlushMessage.IDENTIFIER:
+            // Skip sync/flush in COPY_IN. This is consistent with real PostgreSQL which also does
+            // this to accommodate clients that do not check what type of statement they sent in an
+            // ExecuteMessage, and instead always blindly send a flush/sync after each execute.
+            return SkipMessage.createForValidStream(connection);
+          default:
+            // Skip other unexpected messages and throw an exception to fail the copy operation.
+            validMessage = false;
+            SkipMessage.createForInvalidStream(connection);
+            throw new IllegalStateException(
+                String.format(
+                    "Expected CopyData ('d'), CopyDone ('c') or CopyFail ('f') messages, got: '%c'",
+                    nextMsg));
+        }
+      } else {
+        switch (nextMsg) {
+          case QueryMessage.IDENTIFIER:
+            return new QueryMessage(connection);
+          case ParseMessage.IDENTIFIER:
+            return new ParseMessage(connection);
+          case BindMessage.IDENTIFIER:
+            return new BindMessage(connection);
+          case DescribeMessage.IDENTIFIER:
+            return new DescribeMessage(connection);
+          case ExecuteMessage.IDENTIFIER:
+            return new ExecuteMessage(connection);
+          case CloseMessage.IDENTIFIER:
+            return new CloseMessage(connection);
+          case TerminateMessage.IDENTIFIER:
+            return new TerminateMessage(connection);
+          case FunctionCallMessage.IDENTIFIER:
+            return new FunctionCallMessage(connection);
+          case FlushMessage.IDENTIFIER:
+            return new FlushMessage(connection);
+          case SyncMessage.IDENTIFIER:
+            return new SyncMessage(connection);
+          case CopyDoneMessage.IDENTIFIER:
+          case CopyDataMessage.IDENTIFIER:
+          case CopyFailMessage.IDENTIFIER:
+            // Silently skip COPY messages in non-COPY mode. This is consistent with the PG wire
+            // protocol. If we continue to receive COPY messages while in non-COPY mode, we'll
+            // terminate the connection to prevent the server from being flooded with invalid
+            // messages.
+            validMessage = false;
+            // Note: The stream itself is still valid as we received a message that we recognized.
+            return SkipMessage.createForValidStream(connection);
+          default:
+            throw new IllegalStateException(String.format("Unknown message: %c", nextMsg));
+        }
+      }
+    } finally {
+      if (validMessage) {
+        connection.clearInvalidMessageCount();
+      } else {
+        connection.increaseInvalidMessageCount();
+        if (connection.getInvalidMessageCount() > MAX_INVALID_MESSAGE_COUNT) {
+          new ErrorResponse(
+                  connection,
+                  PGException.newBuilder(
+                          String.format(
+                              "Received %d invalid/unexpected messages. Last received message: '%c'",
+                              connection.getInvalidMessageCount(), nextMsg))
+                      .setSQLState(SQLState.ProtocolViolation)
+                      .setSeverity(Severity.FATAL)
+                      .build())
+              .send();
+          connection.setStatus(ConnectionStatus.TERMINATED);
+        }
+      }
+    }
+  }
+
+  /** Reads the next control message identifier. */
+  private char readNextMsgIdentifier(ConnectionHandler connection) throws IOException {
+    DataInputStream inputStream = connection.getConnectionMetadata().getInputStream();
+    return (char) inputStream.readUnsignedByte();
+  }
+
+  /**
+   * Read the format codes from the current message (useful for both input and output format codes).
+   *
+   * @param input The data stream for the connection.
+   * @return A list of format codes.
+   * @throws Exception If reading fails in any way.
+   */
+  static short[] getFormatCodes(DataInputStream input) throws Exception {
+    short numberOfFormatCodes = input.readShort();
+    if (numberOfFormatCodes == 0) {
+      return NO_FORMAT_CODES;
+    }
+    short[] formatCodes = new short[numberOfFormatCodes];
+    for (int i = 0; i < numberOfFormatCodes; i++) {
+      formatCodes[i] = input.readShort();
+    }
+    return formatCodes;
+  }
+}
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/WireMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/WireMessage.java
index 11dd21be..70e1506b 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/WireMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/WireMessage.java
@@ -32,14 +32,16 @@ public abstract class WireMessage {
 
   private static final Logger logger = Logger.getLogger(WireMessage.class.getName());
 
-  protected int length;
-  protected DataInputStream inputStream;
-  protected DataOutputStream outputStream;
-  protected ConnectionHandler connection;
+  protected final int length;
+  protected final MessageReader messageReader;
+  protected final DataInputStream inputStream;
+  protected final DataOutputStream outputStream;
+  protected final ConnectionHandler connection;
 
   public WireMessage(ConnectionHandler connection, int length) {
     Preconditions.checkArgument(length >= 4);
     this.connection = connection;
+    this.messageReader = connection.getServer().getMessageReader();
     this.inputStream = connection.getConnectionMetadata().getInputStream();
     this.outputStream = connection.getConnectionMetadata().getOutputStream();
     this.length = length;
@@ -210,6 +212,6 @@ public abstract class WireMessage {
    * setting for {@link ConnectionHandler}.
    */
   public void nextHandler() throws Exception {
-    this.connection.setMessageState(ControlMessage.create(this.connection));
+    this.connection.setMessageState(this.messageReader.create(this.connection));
   }
 }
diff --git a/wrappers/golang/go.mod b/wrappers/golang/go.mod
index 9f6337af..0629e3e9 100644
--- a/wrappers/golang/go.mod
+++ b/wrappers/golang/go.mod
@@ -5,7 +5,7 @@ go 1.23.0
 toolchain go1.24.2
 
 require (
-	github.com/docker/docker v28.0.4+incompatible
+	github.com/docker/docker v28.1.1+incompatible
 	github.com/testcontainers/testcontainers-go v0.36.0
 	golang.org/x/oauth2 v0.29.0
 )
@@ -13,7 +13,7 @@ require (
 require (
 	cloud.google.com/go/compute/metadata v0.3.0 // indirect
 	dario.cat/mergo v1.0.1 // indirect
-	github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24 // indirect
+	github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 // indirect
 	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
 	github.com/Microsoft/go-winio v0.6.2 // indirect
 	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
@@ -31,13 +31,14 @@ require (
 	github.com/go-ole/go-ole v1.2.6 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/google/uuid v1.6.0 // indirect
-	github.com/klauspost/compress v1.17.4 // indirect
+	github.com/klauspost/compress v1.18.0 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
 	github.com/magiconair/properties v1.8.9 // indirect
 	github.com/moby/docker-image-spec v1.3.1 // indirect
+	github.com/moby/go-archive v0.1.0 // indirect
 	github.com/moby/patternmatcher v0.6.0 // indirect
-	github.com/moby/sys/sequential v0.5.0 // indirect
-	github.com/moby/sys/user v0.1.0 // indirect
+	github.com/moby/sys/sequential v0.6.0 // indirect
+	github.com/moby/sys/user v0.4.0 // indirect
 	github.com/moby/sys/userns v0.1.0 // indirect
 	github.com/moby/term v0.5.0 // indirect
 	github.com/morikuni/aec v1.0.0 // indirect
diff --git a/wrappers/golang/go.sum b/wrappers/golang/go.sum
index 4ed0f657..f402d3fb 100644
--- a/wrappers/golang/go.sum
+++ b/wrappers/golang/go.sum
@@ -6,6 +6,7 @@ dario.cat/mergo v1.0.1 h1:Ra4+bf83h2ztPIQYNP99R6m+Y7KfnARDfID+a+vLl4s=
 dario.cat/mergo v1.0.1/go.mod h1:uNxQE+84aUszobStD9th8a29P2fMDhsBdgRYvZOxGmk=
 github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24 h1:bvDV9vkmnHYOMsOr4WLk+Vo07yKIzd94sVoIqshQ4bU=
 github.com/AdaLogics/go-fuzz-headers v0.0.0-20230811130428-ced1acdcaa24/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8=
+github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8=
 github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 h1:UQHMgLO+TxOElx5B5HZ4hJQsoJ/PvUvKRhJHDQXO8P8=
 github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E=
 github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY=
@@ -31,6 +32,8 @@ github.com/docker/docker v28.0.2+incompatible h1:9BILleFwug5FSSqWBgVevgL3ewDJfWW
 github.com/docker/docker v28.0.2+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/docker v28.0.4+incompatible h1:JNNkBctYKurkw6FrHfKqY0nKIDf5nrbxjVBtS+cdcok=
 github.com/docker/docker v28.0.4+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
+github.com/docker/docker v28.1.1+incompatible h1:49M11BFLsVO1gxY9UX9p/zwkE/rswggs8AdFmXQw51I=
+github.com/docker/docker v28.1.1+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/go-connections v0.5.0 h1:USnMq7hx7gwdVZq1L49hLXaFtUdTADjXGp+uj1Br63c=
 github.com/docker/go-connections v0.5.0/go.mod h1:ov60Kzw0kKElRwhNs9UlUHAE/F9Fe6GLaXnqyDdmEXc=
 github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
@@ -61,6 +64,8 @@ github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.17.4 h1:Ej5ixsIri7BrIjBkRZLTo6ghwrEtHFk7ijlczPW4fZ4=
 github.com/klauspost/compress v1.17.4/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM=
+github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
+github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
@@ -73,12 +78,18 @@ github.com/magiconair/properties v1.8.9 h1:nWcCbLq1N2v/cpNsy5WvQ37Fb+YElfq20WJ/a
 github.com/magiconair/properties v1.8.9/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0=
 github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0=
 github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo=
+github.com/moby/go-archive v0.1.0 h1:Kk/5rdW/g+H8NHdJW2gsXyZ7UnzvJNOy6VKJqueWdcQ=
+github.com/moby/go-archive v0.1.0/go.mod h1:G9B+YoujNohJmrIYFBpSd54GTUB4lt9S+xVQvsJyFuo=
 github.com/moby/patternmatcher v0.6.0 h1:GmP9lR19aU5GqSSFko+5pRqHi+Ohk1O69aFiKkVGiPk=
 github.com/moby/patternmatcher v0.6.0/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YOUYxkhApJIxc=
 github.com/moby/sys/sequential v0.5.0 h1:OPvI35Lzn9K04PBbCLW0g4LcFAJgHsvXsRyewg5lXtc=
 github.com/moby/sys/sequential v0.5.0/go.mod h1:tH2cOOs5V9MlPiXcQzRC+eEyab644PWKGRYaaV5ZZlo=
+github.com/moby/sys/sequential v0.6.0 h1:qrx7XFUd/5DxtqcoH1h438hF5TmOvzC/lspjy7zgvCU=
+github.com/moby/sys/sequential v0.6.0/go.mod h1:uyv8EUTrca5PnDsdMGXhZe6CCe8U/UiTWd+lL+7b/Ko=
 github.com/moby/sys/user v0.1.0 h1:WmZ93f5Ux6het5iituh9x2zAG7NFY9Aqi49jjE1PaQg=
 github.com/moby/sys/user v0.1.0/go.mod h1:fKJhFOnsCN6xZ5gSfbM6zaHGgDJMrqt9/reuj4T7MmU=
+github.com/moby/sys/user v0.4.0 h1:jhcMKit7SA80hivmFJcbB1vqmw//wU61Zdui2eQXuMs=
+github.com/moby/sys/user v0.4.0/go.mod h1:bG+tYYYJgaMtRKgEmuueC0hJEAZWwtIbZTB+85uoHjs=
 github.com/moby/sys/userns v0.1.0 h1:tVLXkFOxVu9A64/yh59slHVv9ahO9UIev4JZusOLG/g=
 github.com/moby/sys/userns v0.1.0/go.mod h1:IHUYgu/kao6N8YZlp9Cf444ySSvCmDlmzUcYfDHOl28=
 github.com/moby/term v0.5.0 h1:xt8Q1nalod/v7BqbG21f8mQPqH+xAaC9C3N3wfWbVP0=


Test patch:
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java
index bae11d0d..ab9ebbff 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java
@@ -57,7 +57,7 @@ import com.google.cloud.spanner.pgadapter.session.SessionState;
 import com.google.cloud.spanner.pgadapter.utils.ClientAutoDetector.WellKnownClient;
 import com.google.cloud.spanner.pgadapter.utils.Metrics;
 import com.google.cloud.spanner.pgadapter.utils.MutationWriter;
-import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage;
+import com.google.cloud.spanner.pgadapter.wireprotocol.MessageReader;
 import com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage;
 import com.google.cloud.spanner.pgadapter.wireprotocol.WireMessage;
 import com.google.common.collect.ImmutableList;
@@ -477,15 +477,16 @@ public class StatementTest {
     byte[] value = Bytes.concat(messageMetadata, payload.getBytes());
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
 
-    when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
+    when(server.getMessageReader()).thenReturn(new MessageReader(mock(OptionsMetadata.class)));
     when(connectionHandler.getServer()).thenReturn(server);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
     when(server.getOptions()).thenReturn(options);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(QueryMessage.class, message.getClass());
     SimpleQueryStatement simpleQueryStatement = ((QueryMessage) message).getSimpleQueryStatement();
 
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ControlMessageTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ControlMessageTest.java
index b429c7f0..d5c7c567 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ControlMessageTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ControlMessageTest.java
@@ -20,7 +20,6 @@ import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import com.google.cloud.spanner.connection.AbstractStatementParser.StatementType;
-import com.google.cloud.spanner.connection.Connection;
 import com.google.cloud.spanner.connection.StatementResult;
 import com.google.cloud.spanner.connection.StatementResult.ResultType;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
@@ -65,7 +64,6 @@ public final class ControlMessageTest {
   @Mock private ExtendedQueryProtocolHandler extendedQueryProtocolHandler;
   @Mock private IntermediateStatement intermediateStatement;
   @Mock private ConnectionMetadata connectionMetadata;
-  @Mock private Connection connection;
 
   @Test
   public void testInsertResult() throws Exception {
@@ -102,7 +100,7 @@ public final class ControlMessageTest {
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
 
-    ControlMessage controlMessage = ControlMessage.create(connectionHandler);
+    ControlMessage controlMessage = server.getMessageReader().create(connectionHandler);
     controlMessage.sendSpannerResult(intermediateStatement, QueryMode.SIMPLE, 0L);
 
     DataInputStream outputReader =
@@ -126,6 +124,7 @@ public final class ControlMessageTest {
         new DataInputStream(
             new ByteArrayInputStream(new byte[] {(byte) QUERY_IDENTIFIER, 0, 0, 0, 5, 0}));
 
+    when(connectionHandler.getServer()).thenReturn(mock(ProxyServer.class));
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
     when(connectionHandler.getExtendedQueryProtocolHandler())
@@ -157,6 +156,7 @@ public final class ControlMessageTest {
   public void testSendNoRowsAsResultSetFails() {
     OpenTelemetry otel = OpenTelemetry.noop();
     Tracer tracer = otel.getTracer("test");
+    when(connectionHandler.getServer()).thenReturn(mock(ProxyServer.class));
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ProtocolTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ProtocolTest.java
index 273145ad..fc83f50d 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ProtocolTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ProtocolTest.java
@@ -15,7 +15,7 @@
 package com.google.cloud.spanner.pgadapter.wireprotocol;
 
 import static com.google.cloud.spanner.pgadapter.statements.IntermediatePortalStatement.NO_FORMAT_CODES;
-import static com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage.MAX_INVALID_MESSAGE_COUNT;
+import static com.google.cloud.spanner.pgadapter.wireprotocol.MessageReader.MAX_INVALID_MESSAGE_COUNT;
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
@@ -145,6 +145,7 @@ public class ProtocolTest {
 
     when(connectionHandler.getServer()).thenReturn(server);
     when(server.getOptions()).thenReturn(options);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(options.requiresMatcher()).thenReturn(false);
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
@@ -156,7 +157,7 @@ public class ProtocolTest {
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(QueryMessage.class, message.getClass());
     assertEquals(expectedSQL, ((QueryMessage) message).getStatement().getSql());
 
@@ -179,11 +180,12 @@ public class ProtocolTest {
         .thenReturn(extendedQueryProtocolHandler);
     when(connectionHandler.getServer()).thenReturn(server);
     when(server.getOptions()).thenReturn(options);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(QueryMessage.class, message.getClass());
     assertNotNull(((QueryMessage) message).getSimpleQueryStatement());
     assertEquals(expectedSQL, ((QueryMessage) message).getStatement().getSql());
@@ -197,13 +199,15 @@ public class ProtocolTest {
 
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
 
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
 
-    assertThrows(IOException.class, () -> ControlMessage.create(connectionHandler));
+    assertThrows(IOException.class, () -> server.getMessageReader().create(connectionHandler));
   }
 
   @Test
@@ -242,6 +246,7 @@ public class ProtocolTest {
 
     when(connectionHandler.getServer()).thenReturn(server);
     when(server.getOptions()).thenReturn(options);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
@@ -249,7 +254,7 @@ public class ProtocolTest {
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(ParseMessage.class, message.getClass());
     assertEquals(expectedMessageName, ((ParseMessage) message).getName());
     assertEquals(expectedSQL, ((ParseMessage) message).getStatement().getSql());
@@ -306,6 +311,7 @@ public class ProtocolTest {
 
     when(connectionHandler.getServer()).thenReturn(server);
     when(server.getOptions()).thenReturn(options);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
@@ -313,7 +319,7 @@ public class ProtocolTest {
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(ParseMessage.class, message.getClass());
     assertEquals(expectedMessageName, ((ParseMessage) message).getName());
     assertEquals(expectedSQL, ((ParseMessage) message).getStatement().getSql());
@@ -371,6 +377,7 @@ public class ProtocolTest {
 
     when(connectionHandler.getServer()).thenReturn(server);
     when(server.getOptions()).thenReturn(options);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
@@ -378,7 +385,7 @@ public class ProtocolTest {
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(ParseMessage.class, message.getClass());
     assertEquals(expectedMessageName, ((ParseMessage) message).getName());
     assertEquals(expectedSQL, ((ParseMessage) message).getStatement().getSql());
@@ -421,6 +428,7 @@ public class ProtocolTest {
 
     when(connectionHandler.getServer()).thenReturn(server);
     when(server.getOptions()).thenReturn(options);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
@@ -428,7 +436,7 @@ public class ProtocolTest {
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(ParseMessage.class, message.getClass());
     assertEquals(expectedMessageName, ((ParseMessage) message).getName());
     assertEquals(expectedSQL, ((ParseMessage) message).getStatement().getSql());
@@ -479,6 +487,7 @@ public class ProtocolTest {
 
     when(connectionHandler.getServer()).thenReturn(server);
     when(server.getOptions()).thenReturn(options);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
@@ -486,7 +495,7 @@ public class ProtocolTest {
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
 
     when(connectionHandler.hasStatement(anyString())).thenReturn(true);
     assertThrows(IllegalStateException.class, message::send);
@@ -525,6 +534,7 @@ public class ProtocolTest {
 
     when(connectionHandler.getServer()).thenReturn(server);
     when(server.getOptions()).thenReturn(options);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
@@ -532,7 +542,7 @@ public class ProtocolTest {
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
 
     when(connectionHandler.hasStatement(anyString())).thenReturn(true);
     assertThrows(IllegalStateException.class, message::send);
@@ -572,6 +582,7 @@ public class ProtocolTest {
 
     when(connectionHandler.getServer()).thenReturn(server);
     when(server.getOptions()).thenReturn(options);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
@@ -579,7 +590,7 @@ public class ProtocolTest {
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
 
     message.send();
   }
@@ -621,6 +632,7 @@ public class ProtocolTest {
             parameter,
             resultCodesCount);
 
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionHandler.getStatement(anyString())).thenReturn(intermediatePreparedStatement);
     when(intermediatePreparedStatement.createPortal(anyString(), any(), any(), any()))
         .thenReturn(intermediatePortalStatement);
@@ -642,8 +654,9 @@ public class ProtocolTest {
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(BindMessage.class, message.getClass());
     assertEquals(expectedPortalName, ((BindMessage) message).getPortalName());
     assertEquals(expectedStatementName, ((BindMessage) message).getStatementName());
@@ -720,6 +733,7 @@ public class ProtocolTest {
 
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
 
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
@@ -728,8 +742,9 @@ public class ProtocolTest {
     when(connectionHandler.getStatement(anyString())).thenReturn(intermediatePreparedStatement);
     when(intermediatePreparedStatement.createPortal(anyString(), any(), any(), any()))
         .thenReturn(intermediatePortalStatement);
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(BindMessage.class, message.getClass());
     assertEquals(expectedPortalName, ((BindMessage) message).getPortalName());
     assertEquals(expectedStatementName, ((BindMessage) message).getStatementName());
@@ -795,6 +810,7 @@ public class ProtocolTest {
 
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
 
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
@@ -803,8 +819,9 @@ public class ProtocolTest {
     when(connectionHandler.getStatement(anyString())).thenReturn(intermediatePreparedStatement);
     when(intermediatePreparedStatement.createPortal(anyString(), any(), any(), any()))
         .thenReturn(intermediatePortalStatement);
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(BindMessage.class, message.getClass());
     assertEquals(expectedPortalName, ((BindMessage) message).getPortalName());
     assertEquals(expectedStatementName, ((BindMessage) message).getStatementName());
@@ -828,6 +845,7 @@ public class ProtocolTest {
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionHandler.getPortal(anyString())).thenReturn(intermediatePortalStatement);
     when(intermediatePortalStatement.getSql()).thenReturn("select * from foo");
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
@@ -835,8 +853,9 @@ public class ProtocolTest {
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(DescribeMessage.class, message.getClass());
     assertEquals(expectedStatementName, ((DescribeMessage) message).getName());
     assertEquals("select * from foo", ((DescribeMessage) message).getSql());
@@ -867,14 +886,16 @@ public class ProtocolTest {
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionHandler.getStatement(anyString())).thenReturn(intermediatePreparedStatement);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(DescribeMessage.class, message.getClass());
     assertEquals(expectedStatementName, ((DescribeMessage) message).getName());
 
@@ -903,6 +924,7 @@ public class ProtocolTest {
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionHandler.getStatement(anyString())).thenReturn(intermediatePreparedStatement);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
@@ -912,8 +934,9 @@ public class ProtocolTest {
     when(intermediatePreparedStatement.hasException()).thenReturn(true);
     when(intermediatePreparedStatement.getException())
         .thenReturn(PGExceptionFactory.newPGException("test error", SQLState.InternalError));
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(DescribeMessage.class, message.getClass());
     DescribeMessage describeMessage = (DescribeMessage) message;
 
@@ -938,6 +961,8 @@ public class ProtocolTest {
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
     when(connectionHandler.getPortal(anyString())).thenReturn(intermediatePortalStatement);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
@@ -946,7 +971,7 @@ public class ProtocolTest {
         .thenReturn(extendedQueryProtocolHandler);
     when(extendedQueryProtocolHandler.getBackendConnection()).thenReturn(backendConnection);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(ExecuteMessage.class, message.getClass());
     assertEquals(expectedStatementName, ((ExecuteMessage) message).getName());
     assertEquals(totalRows, ((ExecuteMessage) message).getMaxRows());
@@ -985,6 +1010,7 @@ public class ProtocolTest {
 
     PGException testException =
         PGExceptionFactory.newPGException("test error", SQLState.SyntaxError);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(intermediatePortalStatement.hasException()).thenReturn(true);
     when(intermediatePortalStatement.getException()).thenReturn(testException);
     when(connectionHandler.getWellKnownClient()).thenReturn(WellKnownClient.UNSPECIFIED);
@@ -995,8 +1021,9 @@ public class ProtocolTest {
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
     when(extendedQueryProtocolHandler.getBackendConnection()).thenReturn(backendConnection);
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(ExecuteMessage.class, message.getClass());
     assertEquals(expectedStatementName, ((ExecuteMessage) message).getName());
     assertEquals(totalRows, ((ExecuteMessage) message).getMaxRows());
@@ -1028,12 +1055,14 @@ public class ProtocolTest {
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionHandler.getPortal(anyString())).thenReturn(intermediatePortalStatement);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(CloseMessage.class, message.getClass());
     assertEquals(expectedStatementName, ((CloseMessage) message).getName());
     assertEquals(expectedType, ((CloseMessage) message).getType());
@@ -1066,12 +1095,14 @@ public class ProtocolTest {
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
     when(connectionHandler.getStatement(anyString())).thenReturn(intermediatePortalStatement);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(CloseMessage.class, message.getClass());
     assertEquals(expectedStatementName, ((CloseMessage) message).getName());
     assertEquals(expectedType, ((CloseMessage) message).getType());
@@ -1099,6 +1130,8 @@ public class ProtocolTest {
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
     when(connectionHandler.getTraceConnectionId()).thenReturn(UUID.randomUUID());
     ExtendedQueryProtocolHandler extendedQueryProtocolHandler =
         new ExtendedQueryProtocolHandler(connectionHandler, backendConnection);
@@ -1111,7 +1144,7 @@ public class ProtocolTest {
         .thenReturn(extendedQueryProtocolHandler);
     when(backendConnection.getConnectionState()).thenReturn(ConnectionState.IDLE);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(message.getClass(), SyncMessage.class);
 
     message.send();
@@ -1135,6 +1168,7 @@ public class ProtocolTest {
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionHandler.getTraceConnectionId()).thenReturn(UUID.randomUUID());
     ExtendedQueryProtocolHandler extendedQueryProtocolHandler =
         new ExtendedQueryProtocolHandler(connectionHandler, backendConnection);
@@ -1142,12 +1176,13 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getServer()).thenReturn(server);
 
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
     when(backendConnection.getConnectionState()).thenReturn(ConnectionState.TRANSACTION);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(SyncMessage.class, message.getClass());
 
     message.send();
@@ -1170,13 +1205,15 @@ public class ProtocolTest {
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
     DataOutputStream outputStream = mock(DataOutputStream.class);
 
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(message.getClass(), FlushMessage.class);
 
     message.send();
@@ -1196,13 +1233,15 @@ public class ProtocolTest {
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
     DataOutputStream outputStream = mock(DataOutputStream.class);
 
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(FlushMessage.class, message.getClass());
 
     message.send();
@@ -1239,6 +1278,7 @@ public class ProtocolTest {
     when(backendConnection.getMetrics()).thenReturn(new Metrics(OpenTelemetry.noop()));
     OptionsMetadata options = mock(OptionsMetadata.class);
     when(server.getOptions()).thenReturn(options);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
 
@@ -1247,7 +1287,7 @@ public class ProtocolTest {
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(QueryMessage.class, message.getClass());
     assertEquals(expectedSQL, ((QueryMessage) message).getStatement().getSql());
 
@@ -1286,8 +1326,10 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(TerminateMessage.class, message.getClass());
   }
 
@@ -1299,8 +1341,10 @@ public class ProtocolTest {
 
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
 
-    assertThrows(IllegalStateException.class, () -> ControlMessage.create(connectionHandler));
+    assertThrows(
+        IllegalStateException.class, () -> server.getMessageReader().create(connectionHandler));
   }
 
   @Test
@@ -1318,11 +1362,13 @@ public class ProtocolTest {
     when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.COPY_IN);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
 
     MutationWriter mw = mock(MutationWriter.class);
     when(copyStatement.getMutationWriter()).thenReturn(mw);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     assertEquals(CopyDataMessage.class, message.getClass());
     assertArrayEquals(payload, ((CopyDataMessage) message).getPayload());
 
@@ -1346,8 +1392,10 @@ public class ProtocolTest {
     when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.COPY_IN);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
     // This should be a no-op.
     message.sendPayload();
   }
@@ -1357,6 +1405,7 @@ public class ProtocolTest {
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.COPY_IN);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
 
     byte[] messageMetadata = {'d'};
     byte[] payload1 = "1\t'one'\n2\t".getBytes();
@@ -1382,10 +1431,11 @@ public class ProtocolTest {
     when(connectionHandler.getActiveCopyStatement()).thenReturn(copyStatement);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getServer()).thenReturn(server);
 
     {
       when(connectionMetadata.getInputStream()).thenReturn(inputStream1);
-      WireMessage message = ControlMessage.create(connectionHandler);
+      WireMessage message = server.getMessageReader().create(connectionHandler);
       assertEquals(CopyDataMessage.class, message.getClass());
       assertArrayEquals(payload1, ((CopyDataMessage) message).getPayload());
       CopyDataMessage copyDataMessage = (CopyDataMessage) message;
@@ -1393,7 +1443,7 @@ public class ProtocolTest {
     }
     {
       when(connectionMetadata.getInputStream()).thenReturn(inputStream2);
-      WireMessage message = ControlMessage.create(connectionHandler);
+      WireMessage message = server.getMessageReader().create(connectionHandler);
       assertEquals(CopyDataMessage.class, message.getClass());
       assertArrayEquals(payload2, ((CopyDataMessage) message).getPayload());
       CopyDataMessage copyDataMessage = (CopyDataMessage) message;
@@ -1401,7 +1451,7 @@ public class ProtocolTest {
     }
     {
       when(connectionMetadata.getInputStream()).thenReturn(inputStream3);
-      WireMessage message = ControlMessage.create(connectionHandler);
+      WireMessage message = server.getMessageReader().create(connectionHandler);
       assertEquals(CopyDataMessage.class, message.getClass());
       assertArrayEquals(payload3, ((CopyDataMessage) message).getPayload());
       CopyDataMessage copyDataMessage = (CopyDataMessage) message;
@@ -1427,8 +1477,10 @@ public class ProtocolTest {
     when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.COPY_IN);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
 
     assertEquals(CopyDoneMessage.class, message.getClass());
     CopyDoneMessage messageSpy = (CopyDoneMessage) spy(message);
@@ -1459,8 +1511,10 @@ public class ProtocolTest {
     when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.COPY_IN);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
 
     assertEquals(CopyFailMessage.class, message.getClass());
     assertEquals(expectedErrorMessage, ((CopyFailMessage) message).getErrorMessage());
@@ -1514,8 +1568,10 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    WireMessage message = ControlMessage.create(connectionHandler);
+    WireMessage message = server.getMessageReader().create(connectionHandler);
 
     assertEquals(FunctionCallMessage.class, message.getClass());
     assertThrows(IllegalStateException.class, message::send);
@@ -1564,6 +1620,7 @@ public class ProtocolTest {
     when(sessionState.get(null, "server_version")).thenReturn(serverVersionSetting);
     when(backendConnection.getSessionState()).thenReturn(sessionState);
     when(server.getOptions()).thenReturn(options);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
     when(options.shouldAuthenticate()).thenReturn(false);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
@@ -1664,6 +1721,8 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
 
     WireMessage message = BootstrapMessage.create(connectionHandler);
     assertEquals(CancelMessage.class, message.getClass());
@@ -1694,6 +1753,7 @@ public class ProtocolTest {
     when(options.getSslMode()).thenReturn(SslMode.Enable);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
 
     WireMessage message = BootstrapMessage.create(connectionHandler);
     assertEquals(SSLMessage.class, message.getClass());
@@ -1724,6 +1784,7 @@ public class ProtocolTest {
     when(options.getSslMode()).thenReturn(SslMode.Disable);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
 
     WireMessage message = BootstrapMessage.create(connectionHandler);
     assertEquals(SSLMessage.class, message.getClass());
@@ -1747,6 +1808,8 @@ public class ProtocolTest {
     when(intermediatePortalStatement.containsResultSet()).thenReturn(true);
     when(intermediatePortalStatement.describeAsync(backendConnection))
         .thenReturn(SettableFuture.create());
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
 
     DescribeMessage describeMessage =
         new DescribeMessage(connectionHandler, ManuallyCreatedToken.MANUALLY_CREATED_TOKEN);
@@ -1768,6 +1831,8 @@ public class ProtocolTest {
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
 
     SkipMessage message = SkipMessage.createForValidStream(connectionHandler);
     message.send();
@@ -1791,8 +1856,10 @@ public class ProtocolTest {
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.COPY_IN);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    ControlMessage message = ControlMessage.create(connectionHandler);
+    ControlMessage message = server.getMessageReader().create(connectionHandler);
 
     assertEquals(SkipMessage.class, message.getClass());
     // Verify that nothing was written to the output.
@@ -1811,8 +1878,10 @@ public class ProtocolTest {
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.COPY_IN);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    ControlMessage message = ControlMessage.create(connectionHandler);
+    ControlMessage message = server.getMessageReader().create(connectionHandler);
 
     assertEquals(SkipMessage.class, message.getClass());
     // Verify that nothing was written to the output.
@@ -1833,8 +1902,10 @@ public class ProtocolTest {
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.AUTHENTICATED);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    ControlMessage message = ControlMessage.create(connectionHandler);
+    ControlMessage message = server.getMessageReader().create(connectionHandler);
 
     assertEquals(SkipMessage.class, message.getClass());
     // Verify that nothing was written to the output.
@@ -1853,8 +1924,10 @@ public class ProtocolTest {
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.AUTHENTICATED);
+    when(connectionHandler.getServer()).thenReturn(server);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
 
-    ControlMessage message = ControlMessage.create(connectionHandler);
+    ControlMessage message = server.getMessageReader().create(connectionHandler);
 
     assertEquals(SkipMessage.class, message.getClass());
     // Verify that nothing was written to the output.
@@ -1873,8 +1946,10 @@ public class ProtocolTest {
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.AUTHENTICATED);
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
 
-    ControlMessage message = ControlMessage.create(connectionHandler);
+    ControlMessage message = server.getMessageReader().create(connectionHandler);
 
     assertEquals(SkipMessage.class, message.getClass());
     // Verify that nothing was written to the output.
@@ -1908,16 +1983,18 @@ public class ProtocolTest {
     when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.AUTHENTICATED);
     doCallRealMethod().when(connectionHandler).increaseInvalidMessageCount();
     when(connectionHandler.getInvalidMessageCount()).thenCallRealMethod();
+    when(server.getMessageReader()).thenReturn(new MessageReader(options));
+    when(connectionHandler.getServer()).thenReturn(server);
 
     for (int i = 0; i < MAX_INVALID_MESSAGE_COUNT; i++) {
-      ControlMessage message = ControlMessage.create(connectionHandler);
+      ControlMessage message = server.getMessageReader().create(connectionHandler);
       assertEquals(SkipMessage.class, message.getClass());
       // Verify that nothing was written to the output.
       assertEquals(0, result.size());
       verify(connectionHandler, never()).setStatus(ConnectionStatus.TERMINATED);
     }
 
-    ControlMessage.create(connectionHandler);
+    server.getMessageReader().create(connectionHandler);
     verify(connectionHandler).setStatus(ConnectionStatus.TERMINATED);
     byte[] resultBytes = result.toByteArray();
     assertEquals('E', resultBytes[0]);