From 37b21363367f2f5abe970b1a62f6c8fa5554df42 Mon Sep 17 00:00:00 2001
From: Ian Liao <55819364+ian-Liaozy@users.noreply.github.com>
Date: Fri, 14 Nov 2025 19:50:25 +0000
Subject: [PATCH 1/3] Provide more contextual error message for coder

---
 sdks/python/apache_beam/coders/coder_impl.py | 18 ++++++++++++++++--
 1 file changed, 16 insertions(+), 2 deletions(-)

diff --git a/sdks/python/apache_beam/coders/coder_impl.py b/sdks/python/apache_beam/coders/coder_impl.py
index 35d67258b560..1f5ce464e5cf 100644
--- a/sdks/python/apache_beam/coders/coder_impl.py
+++ b/sdks/python/apache_beam/coders/coder_impl.py
@@ -1014,7 +1014,14 @@ class VarIntCoderImpl(StreamCoderImpl):
   A coder for int objects."""
   def encode_to_stream(self, value, out, nested):
     # type: (int, create_OutputStream, bool) -> None
-    out.write_var_int64(value)
+    try:
+      out.write_var_int64(value)
+    except OverflowError as e:
+      raise OverflowError(
+          f"Integer value '{value}' is out of the encodable range for VarIntCoder. "
+          f"This coder is limited to values that fit within a 64-bit signed integer "
+          f"(-(2**63) to 2**63 - 1). Original error: {e}"
+      ) from e
 
   def decode_from_stream(self, in_stream, nested):
     # type: (create_InputStream, bool) -> int
@@ -1036,7 +1043,14 @@ def decode(self, encoded):
   def estimate_size(self, value, nested=False):
     # type: (Any, bool) -> int
     # Note that VarInts are encoded the same way regardless of nesting.
-    return get_varint_size(value)
+    try:
+      return get_varint_size(value)
+    except OverflowError as e:
+      raise OverflowError(
+          f"Cannot estimate size for integer value '{value}'. "
+          f"Value is out of the range for VarIntCoder (64-bit signed integer). "
+          f"Original error: {e}"
+      ) from e
 
 
 class VarInt32CoderImpl(StreamCoderImpl):

From 321cc60e509762a6d5b2194c57fcad8f113d1109 Mon Sep 17 00:00:00 2001
From: Ian Liao <55819364+ian-Liaozy@users.noreply.github.com>
Date: Fri, 14 Nov 2025 20:24:07 +0000
Subject: [PATCH 2/3] Fix formatting issue

---
 sdks/python/apache_beam/coders/coder_impl.py | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/sdks/python/apache_beam/coders/coder_impl.py b/sdks/python/apache_beam/coders/coder_impl.py
index 1f5ce464e5cf..18111a5517bb 100644
--- a/sdks/python/apache_beam/coders/coder_impl.py
+++ b/sdks/python/apache_beam/coders/coder_impl.py
@@ -1020,8 +1020,7 @@ def encode_to_stream(self, value, out, nested):
       raise OverflowError(
           f"Integer value '{value}' is out of the encodable range for VarIntCoder. "
           f"This coder is limited to values that fit within a 64-bit signed integer "
-          f"(-(2**63) to 2**63 - 1). Original error: {e}"
-      ) from e
+          f"(-(2**63) to 2**63 - 1). Original error: {e}") from e
 
   def decode_from_stream(self, in_stream, nested):
     # type: (create_InputStream, bool) -> int
@@ -1049,8 +1048,7 @@ def estimate_size(self, value, nested=False):
       raise OverflowError(
           f"Cannot estimate size for integer value '{value}'. "
           f"Value is out of the range for VarIntCoder (64-bit signed integer). "
-          f"Original error: {e}"
-      ) from e
+          f"Original error: {e}") from e
 
 
 class VarInt32CoderImpl(StreamCoderImpl):

From 0cd749e489820438f2460f97c2ab21a9e97fa2c1 Mon Sep 17 00:00:00 2001
From: Ian Liao <55819364+ian-Liaozy@users.noreply.github.com>
Date: Mon, 17 Nov 2025 17:31:32 +0000
Subject: [PATCH 3/3] Fix lint error

---
 sdks/python/apache_beam/coders/coder_impl.py | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/sdks/python/apache_beam/coders/coder_impl.py b/sdks/python/apache_beam/coders/coder_impl.py
index 18111a5517bb..03514bb50db0 100644
--- a/sdks/python/apache_beam/coders/coder_impl.py
+++ b/sdks/python/apache_beam/coders/coder_impl.py
@@ -1018,9 +1018,10 @@ def encode_to_stream(self, value, out, nested):
       out.write_var_int64(value)
     except OverflowError as e:
       raise OverflowError(
-          f"Integer value '{value}' is out of the encodable range for VarIntCoder. "
-          f"This coder is limited to values that fit within a 64-bit signed integer "
-          f"(-(2**63) to 2**63 - 1). Original error: {e}") from e
+          f"Integer value '{value}' is out of the encodable range for "
+          f"VarIntCoder. This coder is limited to values that fit "
+          f"within a 64-bit signed integer (-(2**63) to 2**63 - 1). "
+          f"Original error: {e}") from e
 
   def decode_from_stream(self, in_stream, nested):
     # type: (create_InputStream, bool) -> int