Repository: GoogleCloudPlatform/pgadapter
Pull request #3159 (2025-04-11 13:07:19)

chore: move control message creation to separate class
Move the logic for creating a control message to a separate MessageReader class. This makes it easier to optimize this specific part of the code for specific use cases, e.g. introducing a busy-wait for messages in low-QPS scenarios.

This change only moves the code into a separate class, but does not make any changes.


Patch:
diff --git a/.github/workflows/cloud-storage-build-and-push-native-image.yaml b/.github/workflows/cloud-storage-build-and-push-native-image.yaml
new file mode 100644
index 00000000..ca54d329
--- /dev/null
+++ b/.github/workflows/cloud-storage-build-and-push-native-image.yaml
@@ -0,0 +1,67 @@
+name: Build and Push native image to Google Cloud Storage
+
+on:
+  # This allows manual activation of this action for testing.
+  workflow_dispatch:
+
+env:
+  GITHUB_SHA: ${{ github.sha }}
+  GITHUB_REF: ${{ github.ref }}
+
+jobs:
+  setup-build-publish-deploy:
+    name: Setup, Build, and Publish
+    runs-on: ubuntu-latest
+    steps:
+
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - id: 'auth'
+        uses: 'google-github-actions/auth@v2'
+        with:
+          credentials_json: '${{ secrets.CLOUD_SPANNER_PG_ADAPTER_SERVICE_ACCOUNT }}'
+
+      # Setup GraalVM
+      - uses: graalvm/setup-graalvm@v1
+        with:
+          java-version: '23'
+          distribution: 'graalvm-community'
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          native-image-job-reports: 'true'
+
+      # Build the assembly and native image
+      - name: Build assembly
+        run: |
+          mvn package -Passembly -DskipTests
+          export VERSION=`echo $GITHUB_REF | awk -F/ '{print $NF}'`
+          echo $VERSION
+          cd "target/pgadapter"
+          native-image \
+            --initialize-at-build-time=com.google.protobuf,com.google.gson,com.google.cloud.spanner.pgadapter.Server \
+            -J-Xmx14g \
+            -H:IncludeResources=".*" \
+            -H:ReflectionConfigurationFiles=../../build/native-image/reflectconfig.json \
+            -jar pgadapter.jar \
+            --no-fallback
+          
+          tar -czvf pgadapter-linux-x64.tar.gz pgadapter
+          cp "pgadapter-linux-x64.tar.gz" "pgadapter-linux-x64-$VERSION.tar.gz"
+          echo "native_image=pgadapter-linux-x64-$VERSION.tar.gz" >> $GITHUB_ENV
+          echo "native_image_current=pgadapter-linux-x64.tar.gz" >> $GITHUB_ENV
+          echo ${{ env.native_image }}
+          echo ${{ env.native_image_current }}
+
+      # Upload the native image to Google Cloud Storage
+      - id: 'upload-versioned-file'
+        uses: 'google-github-actions/upload-cloud-storage@v2'
+        with:
+          path: ${{ env.native_image }}
+          destination: 'pgadapter-jar-releases'
+          parent: false
+      - id: 'overwrite-current-file'
+        uses: 'google-github-actions/upload-cloud-storage@v2'
+        with:
+          path: ${{ env.native_image_current }}
+          destination: 'pgadapter-jar-releases'
+          parent: false
diff --git a/benchmarks/latency-comparison/golang/go.mod b/benchmarks/latency-comparison/golang/go.mod
index 7662f11e..c8fa8048 100644
--- a/benchmarks/latency-comparison/golang/go.mod
+++ b/benchmarks/latency-comparison/golang/go.mod
@@ -6,24 +6,24 @@ 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
 	github.com/testcontainers/testcontainers-go v0.36.0
 	golang.org/x/oauth2 v0.29.0
-	google.golang.org/api v0.228.0
+	google.golang.org/api v0.229.0
 )
 
 require (
 	cel.dev/expr v0.19.2 // indirect
 	cloud.google.com/go v0.120.0 // indirect
-	cloud.google.com/go/auth v0.15.0 // indirect
+	cloud.google.com/go/auth v0.16.0 // indirect
 	cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect
 	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
@@ -85,24 +86,24 @@ require (
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/contrib/detectors/gcp v1.35.0 // indirect
-	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.59.0 // indirect
-	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.59.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.60.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 // indirect
 	go.opentelemetry.io/otel v1.35.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.19.0 // indirect
 	go.opentelemetry.io/otel/metric v1.35.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.35.0 // indirect
 	go.opentelemetry.io/otel/sdk/metric v1.35.0 // indirect
 	go.opentelemetry.io/otel/trace v1.35.0 // indirect
-	golang.org/x/crypto v0.36.0 // indirect
-	golang.org/x/net v0.37.0 // indirect
-	golang.org/x/sync v0.12.0 // indirect
-	golang.org/x/sys v0.31.0 // indirect
-	golang.org/x/text v0.23.0 // indirect
+	golang.org/x/crypto v0.37.0 // indirect
+	golang.org/x/net v0.39.0 // indirect
+	golang.org/x/sync v0.13.0 // indirect
+	golang.org/x/sys v0.32.0 // indirect
+	golang.org/x/text v0.24.0 // indirect
 	golang.org/x/time v0.11.0 // indirect
 	google.golang.org/genproto v0.0.0-20250303144028-a0af3efb3deb // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20250313205543-e70fdf4c4cb4 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20250313205543-e70fdf4c4cb4 // indirect
-	google.golang.org/grpc v1.71.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250414145226-207652e42e2e // indirect
+	google.golang.org/grpc v1.71.1 // indirect
 	google.golang.org/protobuf v1.36.6 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
diff --git a/benchmarks/latency-comparison/golang/go.sum b/benchmarks/latency-comparison/golang/go.sum
index 2d3caac8..e49c537b 100644
--- a/benchmarks/latency-comparison/golang/go.sum
+++ b/benchmarks/latency-comparison/golang/go.sum
@@ -107,6 +107,8 @@ cloud.google.com/go/assuredworkloads v1.9.0/go.mod h1:kFuI1P78bplYtT77Tb1hi0FMxM
 cloud.google.com/go/assuredworkloads v1.10.0/go.mod h1:kwdUQuXcedVdsIaKgKTp9t0UJkE5+PAVNhdQm4ZVq2E=
 cloud.google.com/go/auth v0.15.0 h1:Ly0u4aA5vG/fsSsxu98qCQBemXtAtJf+95z9HK+cxps=
 cloud.google.com/go/auth v0.15.0/go.mod h1:WJDGqZ1o9E9wKIL+IwStfyn/+s59zl4Bi+1KQNVXLZ8=
+cloud.google.com/go/auth v0.16.0 h1:Pd8P1s9WkcrBE2n/PhAwKsdrR35V3Sg2II9B+ndM3CU=
+cloud.google.com/go/auth v0.16.0/go.mod h1:1howDHJ5IETh/LwYs3ZxvlkXF48aSqqJUM+5o02dNOI=
 cloud.google.com/go/auth/oauth2adapt v0.2.7 h1:/Lc7xODdqcEw8IrZ9SvwnlLX6j9FHQM74z6cBk9Rw6M=
 cloud.google.com/go/auth/oauth2adapt v0.2.7/go.mod h1:NTbTTzfvPl1Y3V1nPpOgl2w6d/FjO7NNUQaWSox6ZMc=
 cloud.google.com/go/auth/oauth2adapt v0.2.8 h1:keo8NaayQZ6wimpNSmW5OPc283g65QNIiLpZnkHRbnc=
@@ -631,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=
@@ -713,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=
@@ -968,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=
@@ -1007,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=
@@ -1137,8 +1151,12 @@ go.opentelemetry.io/contrib/detectors/gcp v1.35.0 h1:bGvFt68+KTiAKFlacHW6AhA56GF
 go.opentelemetry.io/contrib/detectors/gcp v1.35.0/go.mod h1:qGWP8/+ILwMRIUf9uIVLloR1uo5ZYAslM4O6OqUi1DA=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.59.0 h1:rgMkmiGfix9vFJDcDi1PK8WEQP4FLQwLDfhp5ZLpFeE=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.59.0/go.mod h1:ijPqXp5P6IRRByFVVg9DY8P5HkxkHE5ARIa+86aXPf4=
+go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.60.0 h1:x7wzEgXfnzJcHDwStJT+mxOz4etr2EcexjqhBvmoakw=
+go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.60.0/go.mod h1:rg+RlpR5dKwaS95IyyZqj5Wd4E13lk/msnTS0Xl9lJM=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.59.0 h1:CV7UdSGJt/Ao6Gp4CXckLxVRRsRgDHoI8XjbL3PDl8s=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.59.0/go.mod h1:FRmFuRJfag1IZ2dPkHnEoSFVgTVPUd2qf5Vi69hLb8I=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 h1:sbiXRNDSWJOTobXh5HyQKjq6wUC5tNybqjIqDpAY4CU=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0/go.mod h1:69uWxva0WgAA/4bu2Yy70SLDBwZXuQ6PbBpbsa5iZrQ=
 go.opentelemetry.io/otel v1.34.0 h1:zRLXxLCgL1WyKsPVrgbSdMN4c0FMkDAskSTQP+0hdUY=
 go.opentelemetry.io/otel v1.34.0/go.mod h1:OWFPOQ+h4G8xpyjgqo4SxJYdDQ/qmRH+wivy7zzx9oI=
 go.opentelemetry.io/otel v1.35.0 h1:xKWKPxrxB6OtMCbmMY021CqC45J+3Onta9MqjhnusiQ=
@@ -1194,6 +1212,8 @@ golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5y
 golang.org/x/crypto v0.0.0-20211108221036-ceb1ce70b4fa/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.36.0 h1:AnAEvhDddvBdpY+uR+MyHmuZzzNqXSe/GvuDeob5L34=
 golang.org/x/crypto v0.36.0/go.mod h1:Y4J0ReaxCR1IMaabaSMugxJES1EpwhBHhv2bDHklZvc=
+golang.org/x/crypto v0.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE=
+golang.org/x/crypto v0.37.0/go.mod h1:vg+k43peMZ0pUMhYmVAWysMK35e6ioLh3wB8ZCAfbVc=
 golang.org/x/exp v0.0.0-20180321215751-8460e604b9de/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20180807140117-3d87b88a115f/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -1311,6 +1331,8 @@ golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc=
 golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns=
 golang.org/x/net v0.37.0 h1:1zLorHbz+LYj7MQlSf1+2tPIIgibq2eL5xkrGk6f+2c=
 golang.org/x/net v0.37.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8=
+golang.org/x/net v0.39.0 h1:ZCu7HMWDxpXpaiKdhzIfaltL9Lp31x/3fCP11bc6/fY=
+golang.org/x/net v0.39.0/go.mod h1:X7NRbYVEA+ewNkCNyJ513WmMdQ3BineSwVtN2zD/d+E=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -1362,6 +1384,8 @@ golang.org/x/sync v0.0.0-20220929204114-8fcdb60fdcc0/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.12.0 h1:MHc5BpPuC30uJk597Ri8TV3CNZcTLu6B6z4lJy+g6Jw=
 golang.org/x/sync v0.12.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
+golang.org/x/sync v0.13.0 h1:AauUjRAJ9OSnvULf/ARrrVywoJDy0YS2AwQ98I37610=
+golang.org/x/sync v0.13.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -1453,6 +1477,8 @@ golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik=
 golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/sys v0.32.0 h1:s77OFDvIQeibCmezSnk/q6iAfkdiQaJi4VzroCFrN20=
+golang.org/x/sys v0.32.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
@@ -1482,6 +1508,8 @@ golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY=
 golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4=
+golang.org/x/text v0.24.0 h1:dd5Bzh4yt5KYA8f9CJHCP4FB4D51c2c6JvN37xJJkJ0=
+golang.org/x/text v0.24.0/go.mod h1:L8rBsPeo2pSS+xqN0d5u2ikmjtmoJbDBT1b7nHvFCdU=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
@@ -1639,6 +1667,8 @@ google.golang.org/api v0.227.0 h1:QvIHF9IuyG6d6ReE+BNd11kIB8hZvjN8Z5xY5t21zYc=
 google.golang.org/api v0.227.0/go.mod h1:EIpaG6MbTgQarWF5xJvX0eOJPK9n/5D4Bynb9j2HXvQ=
 google.golang.org/api v0.228.0 h1:X2DJ/uoWGnY5obVjewbp8icSL5U4FzuCfy9OjbLSnLs=
 google.golang.org/api v0.228.0/go.mod h1:wNvRS1Pbe8r4+IfBIniV8fwCpGwTrYa+kMUDiC5z5a4=
+google.golang.org/api v0.229.0 h1:p98ymMtqeJ5i3lIBMj5MpR9kzIIgzpHHh8vQ+vgAzx8=
+google.golang.org/api v0.229.0/go.mod h1:wyDfmq5g1wYJWn29O22FDWN48P7Xcz0xz+LBpptYvB0=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/appengine v1.5.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
@@ -1790,6 +1820,8 @@ google.golang.org/genproto/googleapis/rpc v0.0.0-20250303144028-a0af3efb3deb h1:
 google.golang.org/genproto/googleapis/rpc v0.0.0-20250303144028-a0af3efb3deb/go.mod h1:LuRYeWDFV6WOn90g357N17oMCaxpgCnbi/44qJvDn2I=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20250313205543-e70fdf4c4cb4 h1:iK2jbkWL86DXjEx0qiHcRE9dE4/Ahua5k6V8OWFb//c=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20250313205543-e70fdf4c4cb4/go.mod h1:LuRYeWDFV6WOn90g357N17oMCaxpgCnbi/44qJvDn2I=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250414145226-207652e42e2e h1:ztQaXfzEXTmCBvbtWYRhJxW+0iJcz2qXfd38/e9l7bA=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250414145226-207652e42e2e/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38=
 google.golang.org/grpc v1.21.1/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM=
@@ -1833,6 +1865,8 @@ google.golang.org/grpc v1.54.0/go.mod h1:PUSEXI6iWghWaB6lXM4knEgpJNu2qUcKfDtNci3
 google.golang.org/grpc v1.56.3/go.mod h1:I9bI3vqKfayGqPUAwGdOSu7kt6oIJLixfffKrpXqQ9s=
 google.golang.org/grpc v1.71.0 h1:kF77BGdPTQ4/JZWMlb9VpJ5pa25aqvVqogsxNHHdeBg=
 google.golang.org/grpc v1.71.0/go.mod h1:H0GRtasmQOh9LkFoCPDu3ZrwUtD1YGE+b2vYBYd/8Ec=
+google.golang.org/grpc v1.71.1 h1:ffsFWr7ygTUscGPI0KKK6TLrGz0476KUvvsbqWK0rPI=
+google.golang.org/grpc v1.71.1/go.mod h1:H0GRtasmQOh9LkFoCPDu3ZrwUtD1YGE+b2vYBYd/8Ec=
 google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0/go.mod h1:6Kw0yEErY5E/yWrBtf03jp27GLLJujG4z/JK95pnjjw=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
@@ -1868,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/build/native-image/README.md b/build/native-image/README.md
new file mode 100644
index 00000000..216c1804
--- /dev/null
+++ b/build/native-image/README.md
@@ -0,0 +1,7 @@
+# Native Image
+
+This directory contains a script for building a binary executable of PGAdapter. This requires
+[GraalVM](https://www.graalvm.org/) to be used for compilation.
+
+Run the `build.sh` file in this directory to generate a binary executable. The binary executable
+will be saved to `target/pgadapter/pgadapter`.
diff --git a/build/native-image/build.sh b/build/native-image/build.sh
new file mode 100644
index 00000000..ee519066
--- /dev/null
+++ b/build/native-image/build.sh
@@ -0,0 +1,9 @@
+mvn clean package -P assembly -DskipTests
+cd target/pgadapter
+native-image \
+  --initialize-at-build-time=com.google.protobuf,com.google.gson,com.google.cloud.spanner.pgadapter.Server \
+  -J-Xmx14g \
+  -H:IncludeResources=".*" \
+  -H:ReflectionConfigurationFiles=../../build/native-image/reflectconfig.json \
+  -jar pgadapter.jar \
+  --no-fallback
diff --git a/build/native-image/reflect-config.json b/build/native-image/reflect-config.json
new file mode 100644
index 00000000..8d116ca9
--- /dev/null
+++ b/build/native-image/reflect-config.json
@@ -0,0 +1,79 @@
+[
+  {
+    "name" : "java.lang.Class",
+    "queryAllDeclaredConstructors" : true,
+    "queryAllPublicConstructors" : true,
+    "queryAllDeclaredMethods" : true,
+    "queryAllPublicMethods" : true,
+    "allDeclaredClasses" : true,
+    "allPublicClasses" : true
+  },
+  {
+    "name" : "sun.misc.Signal",
+    "queryAllDeclaredConstructors" : true,
+    "queryAllPublicConstructors" : true,
+    "queryAllDeclaredMethods" : true,
+    "queryAllPublicMethods" : true,
+    "allDeclaredClasses" : true,
+    "allPublicClasses" : true,
+    "allDeclaredConstructors": true,
+    "allPublicConstructors": true,
+    "allDeclaredMethods": true,
+    "allPublicMethods": true
+  },
+  {
+    "name" : "sun.misc.SignalHandler",
+    "queryAllDeclaredConstructors" : true,
+    "queryAllPublicConstructors" : true,
+    "queryAllDeclaredMethods" : true,
+    "queryAllPublicMethods" : true,
+    "allDeclaredClasses" : true,
+    "allPublicClasses" : true,
+    "allDeclaredConstructors": true,
+    "allPublicConstructors": true,
+    "allDeclaredMethods": true,
+    "allPublicMethods": true
+  },
+  {
+    "name" : "com.google.cloud.spanner.pgadapter.Server",
+    "methods" : [
+      { "name" : "handleTerm" },
+      { "name" : "handleInt" },
+      { "name" : "handleQuit" }
+    ]
+  },
+  {
+    "name" : "com.google.cloud.spanner.pgadapter.logging.StdoutHandler",
+    "queryAllDeclaredConstructors" : true,
+    "queryAllPublicConstructors" : true,
+    "queryAllDeclaredMethods" : true,
+    "queryAllPublicMethods" : true,
+    "allDeclaredClasses" : true,
+    "allPublicClasses" : true,
+    "allDeclaredConstructors": true,
+    "allPublicConstructors": true,
+    "allDeclaredMethods": true,
+    "allPublicMethods": true
+  },
+  {
+    "name" : "com.google.cloud.spanner.pgadapter.logging.StderrHandler",
+    "queryAllDeclaredConstructors" : true,
+    "queryAllPublicConstructors" : true,
+    "queryAllDeclaredMethods" : true,
+    "queryAllPublicMethods" : true,
+    "allDeclaredClasses" : true,
+    "allPublicClasses" : true,
+    "allDeclaredConstructors": true,
+    "allPublicConstructors": true,
+    "allDeclaredMethods": true,
+    "allPublicMethods": true
+  },
+  {
+    "name" : "java.time.Instant",
+    "methods" : [
+      { "name" : "now" },
+      { "name" : "getNano" },
+      { "name" : "getEpochSecond" }
+    ]
+  }
+]
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/dotnet/npgsql-sample/npgsql-sample.csproj b/samples/dotnet/npgsql-sample/npgsql-sample.csproj
index 5b43da5d..1143010a 100644
--- a/samples/dotnet/npgsql-sample/npgsql-sample.csproj
+++ b/samples/dotnet/npgsql-sample/npgsql-sample.csproj
@@ -10,7 +10,7 @@
 
     <ItemGroup>
       <PackageReference Include="Npgsql" Version="9.0.3" />
-      <PackageReference Include="Testcontainers" Version="4.3.0" />
+      <PackageReference Include="Testcontainers" Version="4.4.0" />
     </ItemGroup>
 
 </Project>
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/python/connectorx/requirements.txt b/samples/python/connectorx/requirements.txt
index 6f5520e5..22f81551 100644
--- a/samples/python/connectorx/requirements.txt
+++ b/samples/python/connectorx/requirements.txt
@@ -3,4 +3,4 @@ pandas==2.2.3
 testcontainers~=4.10.0
 requests==2.32.3
 google~=3.0.0
-google.auth~=2.38.0
+google.auth~=2.39.0
diff --git a/samples/python/django/requirements.txt b/samples/python/django/requirements.txt
index 7cb7cba1..e10a9b7c 100644
--- a/samples/python/django/requirements.txt
+++ b/samples/python/django/requirements.txt
@@ -2,6 +2,6 @@ psycopg-binary==3.2.6
 pytz==2025.2
 Django==5.2
 google==3.0.0
-google.auth==2.38.0
+google.auth==2.39.0
 testcontainers==4.10.0
 requests==2.32.3
diff --git a/samples/python/psycopg3/requirements.txt b/samples/python/psycopg3/requirements.txt
index ddee956f..fb5b177a 100644
--- a/samples/python/psycopg3/requirements.txt
+++ b/samples/python/psycopg3/requirements.txt
@@ -2,4 +2,4 @@ psycopg[binary]~=3.2.6
 testcontainers~=4.10.0
 requests==2.32.3
 google~=3.0.0
-google.auth~=2.38.0
\ No newline at end of file
+google.auth~=2.39.0
\ No newline at end of file
diff --git a/samples/python/sqlalchemy2-sample/requirements.txt b/samples/python/sqlalchemy2-sample/requirements.txt
index c0a63919..70905c37 100644
--- a/samples/python/sqlalchemy2-sample/requirements.txt
+++ b/samples/python/sqlalchemy2-sample/requirements.txt
@@ -2,6 +2,6 @@ psycopg[binary]==3.2.6
 pytz==2025.2
 sqlalchemy==2.0.40
 google~=3.0.0
-google.auth~=2.38.0
+google.auth~=2.39.0
 testcontainers~=4.10.0
 requests==2.32.3
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/Server.java b/src/main/java/com/google/cloud/spanner/pgadapter/Server.java
index b0d4e692..066cf681 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/Server.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/Server.java
@@ -63,6 +63,10 @@ public class Server {
 
   private static volatile ShutdownHandler shutdownHandler;
 
+  static {
+    registerSignalHandlers();
+  }
+
   /**
    * Main method for running a Spanner PostgreSQL Adapter {@link Server} as a stand-alone
    * application. Here we call for parameter parsing and start the Proxy Server.
@@ -78,7 +82,6 @@ public class Server {
       // Create a shutdown handler and register signal handlers for the signals that should
       // terminate the server.
       Server.shutdownHandler = proxyServer.getOrCreateShutdownHandler();
-      registerSignalHandlers();
     } catch (Exception e) {
       printError(e, System.err, System.out);
     }
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 7fb6060e..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;
@@ -163,32 +165,35 @@ public abstract class WireMessage {
    */
   public String readString() throws IOException {
     this.inputStream.mark(MARK_READ_LIMIT);
-    int index = 0;
-    while (index < MARK_READ_LIMIT) {
-      byte b = this.inputStream.readByte();
-      if (b == 0) {
-        break;
+    try {
+      int index = 0;
+      while (index < MARK_READ_LIMIT) {
+        byte b = this.inputStream.readByte();
+        if (b == 0) {
+          break;
+        }
+        index++;
+        if (index == MARK_READ_LIMIT) {
+          throw new IOException("No null terminator found");
+        }
       }
-      index++;
-      if (index == MARK_READ_LIMIT) {
-        throw new IOException("No null terminator found");
+      if (index == 0) {
+        // Empty string, we don't have to ready anything.
+        return "";
       }
-    }
-    // Reset the stream to the mark and read the name (if any).
-    this.inputStream.reset();
-    if (index == 0) {
-      // No name, but we still need to skip the null-terminator.
+
+      // Reset the stream to the mark and read the string.
+      this.inputStream.reset();
+      byte[] result = new byte[index];
+      this.inputStream.readFully(result);
+      // Skip the null-terminator.
       //noinspection StatementWithEmptyBody
       while (this.inputStream.skip(1) < 1) {}
-      return "";
+      return new String(result, StandardCharsets.UTF_8);
+    } finally {
+      // Drop the mark.
+      this.inputStream.mark(0);
     }
-
-    byte[] result = new byte[index];
-    this.inputStream.readFully(result);
-    // Skip the null-terminator.
-    //noinspection StatementWithEmptyBody
-    while (this.inputStream.skip(1) < 1) {}
-    return new String(result, StandardCharsets.UTF_8);
   }
 
   /**
@@ -207,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/samples/snippets/dotnet-snippets/sample-tests/sample-tests.csproj b/samples/snippets/dotnet-snippets/sample-tests/sample-tests.csproj
index dfc0696a..e8f33b4d 100644
--- a/samples/snippets/dotnet-snippets/sample-tests/sample-tests.csproj
+++ b/samples/snippets/dotnet-snippets/sample-tests/sample-tests.csproj
@@ -16,7 +16,7 @@
         <PackageReference Include="NUnit" Version="4.3.2" />
         <PackageReference Include="NUnit.Analyzers" Version="4.7.0"/>
         <PackageReference Include="NUnit3TestAdapter" Version="5.0.0"/>
-        <PackageReference Include="Testcontainers" Version="4.3.0" />
+        <PackageReference Include="Testcontainers" Version="4.4.0" />
     </ItemGroup>
 
     <ItemGroup>
diff --git a/src/test/golang/pgadapter_gorm_tests/go.mod b/src/test/golang/pgadapter_gorm_tests/go.mod
index 0dd0b4b8..183364b0 100644
--- a/src/test/golang/pgadapter_gorm_tests/go.mod
+++ b/src/test/golang/pgadapter_gorm_tests/go.mod
@@ -21,8 +21,8 @@ require (
 	github.com/jackc/puddle/v2 v2.2.1 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect
 	github.com/jinzhu/now v1.1.5 // indirect
-	golang.org/x/crypto v0.31.0 // indirect
-	golang.org/x/sync v0.10.0 // indirect
-	golang.org/x/text v0.21.0 // indirect
+	golang.org/x/crypto v0.35.0 // indirect
+	golang.org/x/sync v0.11.0 // indirect
+	golang.org/x/text v0.22.0 // indirect
 	gorm.io/driver/mysql v1.5.6 // indirect
 )
diff --git a/src/test/golang/pgadapter_gorm_tests/go.sum b/src/test/golang/pgadapter_gorm_tests/go.sum
index 34eae0d2..a80ff1c5 100644
--- a/src/test/golang/pgadapter_gorm_tests/go.sum
+++ b/src/test/golang/pgadapter_gorm_tests/go.sum
@@ -210,8 +210,8 @@ golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDf
 golang.org/x/crypto v0.20.0/go.mod h1:Xwo95rrVNIoSMx9wa1JroENMToLWn3RNVrTBpLHgZPQ=
 golang.org/x/crypto v0.22.0/go.mod h1:vr6Su+7cTlO45qkww3VDJlzDn0ctJvRgYbC2NvXHt+M=
 golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8=
-golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
-golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
+golang.org/x/crypto v0.35.0 h1:b15kiHdrGCHrP6LvwaQ3c03kgNhhiMgvlhxHQhmg2Xs=
+golang.org/x/crypto v0.35.0/go.mod h1:dy7dXNW32cAb/6/PRuTNsix8T+vJAqvuIy5Bli/x0YQ=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
 golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc=
 golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
@@ -245,8 +245,8 @@ golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
 golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
-golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
-golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
+golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -275,7 +275,7 @@ golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2/go.mod h1:TeRTkGYfJXctD9OcfyVLyj2J3IxLnKwHJR8f4D8a3YE=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
@@ -292,7 +292,7 @@ golang.org/x/term v0.16.0/go.mod h1:yn7UURbUtPyrVJPGPq404EukNFxcm/foM+bV/bfcDsY=
 golang.org/x/term v0.17.0/go.mod h1:lLRBjIVuehSbZlaOtGMbcMncT+aqLLLmKrsjNrUguwk=
 golang.org/x/term v0.19.0/go.mod h1:2CuTdWZ7KHSQwUzKva0cbMg6q2DMI3Mmxp+gKJbskEk=
 golang.org/x/term v0.20.0/go.mod h1:8UkIAJTvZgivsXaD6/pH6U9ecQzZ45awqEOzuCvwpFY=
-golang.org/x/term v0.27.0/go.mod h1:iMsnZpn0cago0GOrHO2+Y7u7JPn5AylBrcoWkElMTSM=
+golang.org/x/term v0.29.0/go.mod h1:6bl4lRlvVuDgSf3179VpIxBF0o10JUpXWOnI7nErv7s=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
@@ -307,8 +307,8 @@ golang.org/x/text v0.12.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
 golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
-golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
-golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
+golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM=
+golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190425163242-31fd60d6bfdc/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
diff --git a/src/test/golang/pgadapter_pgx5_tests/go.mod b/src/test/golang/pgadapter_pgx5_tests/go.mod
index 9c0fbbd3..fed69cde 100644
--- a/src/test/golang/pgadapter_pgx5_tests/go.mod
+++ b/src/test/golang/pgadapter_pgx5_tests/go.mod
@@ -13,6 +13,6 @@ require (
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgproto3/v2 v2.3.3 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
-	golang.org/x/crypto v0.31.0 // indirect
-	golang.org/x/text v0.21.0 // indirect
+	golang.org/x/crypto v0.35.0 // indirect
+	golang.org/x/text v0.22.0 // indirect
 )
diff --git a/src/test/golang/pgadapter_pgx5_tests/go.sum b/src/test/golang/pgadapter_pgx5_tests/go.sum
index 626aaf44..6bdf10f3 100644
--- a/src/test/golang/pgadapter_pgx5_tests/go.sum
+++ b/src/test/golang/pgadapter_pgx5_tests/go.sum
@@ -130,8 +130,8 @@ golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5y
 golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58=
-golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
-golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
+golang.org/x/crypto v0.35.0 h1:b15kiHdrGCHrP6LvwaQ3c03kgNhhiMgvlhxHQhmg2Xs=
+golang.org/x/crypto v0.35.0/go.mod h1:dy7dXNW32cAb/6/PRuTNsix8T+vJAqvuIy5Bli/x0YQ=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
 golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc=
 golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
@@ -145,7 +145,7 @@ golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug
 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
+golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -172,8 +172,8 @@ golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
-golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
-golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
+golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM=
+golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190425163242-31fd60d6bfdc/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
diff --git a/src/test/golang/pgadapter_pgx_tests/go.mod b/src/test/golang/pgadapter_pgx_tests/go.mod
index fc515e1f..9e736651 100644
--- a/src/test/golang/pgadapter_pgx_tests/go.mod
+++ b/src/test/golang/pgadapter_pgx_tests/go.mod
@@ -14,6 +14,6 @@ require (
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgproto3/v2 v2.3.3 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
-	golang.org/x/crypto v0.31.0 // indirect
-	golang.org/x/text v0.21.0 // indirect
+	golang.org/x/crypto v0.35.0 // indirect
+	golang.org/x/text v0.22.0 // indirect
 )
diff --git a/src/test/golang/pgadapter_pgx_tests/go.sum b/src/test/golang/pgadapter_pgx_tests/go.sum
index 92a8f7a3..ed6ff98f 100644
--- a/src/test/golang/pgadapter_pgx_tests/go.sum
+++ b/src/test/golang/pgadapter_pgx_tests/go.sum
@@ -139,8 +139,8 @@ golang.org/x/crypto v0.13.0/go.mod h1:y6Z2r+Rw4iayiXXAIxJIDAJ1zMW4yaTpebo8fPOliY
 golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU=
 golang.org/x/crypto v0.20.0/go.mod h1:Xwo95rrVNIoSMx9wa1JroENMToLWn3RNVrTBpLHgZPQ=
 golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8=
-golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
-golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
+golang.org/x/crypto v0.35.0 h1:b15kiHdrGCHrP6LvwaQ3c03kgNhhiMgvlhxHQhmg2Xs=
+golang.org/x/crypto v0.35.0/go.mod h1:dy7dXNW32cAb/6/PRuTNsix8T+vJAqvuIy5Bli/x0YQ=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
 golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc=
 golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
@@ -166,7 +166,7 @@ golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
 golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
-golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -186,7 +186,7 @@ golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2/go.mod h1:TeRTkGYfJXctD9OcfyVLyj2J3IxLnKwHJR8f4D8a3YE=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
@@ -196,7 +196,7 @@ golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
 golang.org/x/term v0.12.0/go.mod h1:owVbMEjm3cBLCHdkQu9b1opXd4ETQWc3BhuQGKgXgvU=
 golang.org/x/term v0.17.0/go.mod h1:lLRBjIVuehSbZlaOtGMbcMncT+aqLLLmKrsjNrUguwk=
 golang.org/x/term v0.20.0/go.mod h1:8UkIAJTvZgivsXaD6/pH6U9ecQzZ45awqEOzuCvwpFY=
-golang.org/x/term v0.27.0/go.mod h1:iMsnZpn0cago0GOrHO2+Y7u7JPn5AylBrcoWkElMTSM=
+golang.org/x/term v0.29.0/go.mod h1:6bl4lRlvVuDgSf3179VpIxBF0o10JUpXWOnI7nErv7s=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
@@ -208,8 +208,8 @@ golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
 golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
-golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
-golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
+golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM=
+golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190425163242-31fd60d6bfdc/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
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]);