tolerate integer overflow when encoding varint32

.github/trigger_files/beam_PostCommit_XVR_Direct.json

@@ -1,3 +1,4 @@
 {
-  "https://github.com/apache/beam/pull/32648": "testing Flink 1.19 support"
+  "https://github.com/apache/beam/pull/32648": "testing Flink 1.19 support",
+  "modification": 1
 }

.github/trigger_files/beam_PostCommit_XVR_PythonUsingJavaSQL_Dataflow.json

@@ -1 +1,3 @@
-{}
+{
+  "modification": 1
+}

sdks/python/apache_beam/coders/coder_impl.py

@@ -1002,7 +1002,7 @@ 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 & 0xFFFFFFFF)
+    return get_varint_size(int(value) & 0xFFFFFFFF)
 
 
 class SingletonCoderImpl(CoderImpl):

sdks/python/apache_beam/coders/stream.pxd

@@ -26,7 +26,7 @@ cdef class OutputStream(object):
   cpdef write(self, bytes b, bint nested=*)
   cpdef write_byte(self, unsigned char val)
   cpdef write_var_int64(self, libc.stdint.int64_t v)
-  cpdef write_var_int32(self, libc.stdint.int32_t v)
+  cpdef write_var_int32(self, libc.stdint.int64_t v)
   cpdef write_bigendian_int64(self, libc.stdint.int64_t signed_v)
   cpdef write_bigendian_uint64(self, libc.stdint.uint64_t signed_v)
   cpdef write_bigendian_int32(self, libc.stdint.int32_t signed_v)
@@ -45,7 +45,7 @@ cdef class ByteCountingOutputStream(OutputStream):
 
   cpdef write(self, bytes b, bint nested=*)
   cpdef write_var_int64(self, libc.stdint.int64_t val)
-  cpdef write_var_int32(self, libc.stdint.int32_t val)
+  cpdef write_var_int32(self, libc.stdint.int64_t val)
   cpdef write_byte(self, unsigned char val)
   cpdef write_bigendian_int64(self, libc.stdint.int64_t val)
   cpdef write_bigendian_uint64(self, libc.stdint.uint64_t val)

sdks/python/apache_beam/coders/stream.pyx

@@ -73,9 +73,10 @@ cdef class OutputStream(object):
       if not v:
         break
 
-  cpdef write_var_int32(self, libc.stdint.int32_t signed_v):
+  cpdef write_var_int32(self, libc.stdint.int64_t signed_v):
     """Encode an int using variable-length encoding to a stream."""
-    cdef libc.stdint.int64_t v = signed_v & <libc.stdint.int64_t>(0xFFFFFFFF)
+    # for backward compatibility, input type is int64_t thus tolerates overflow
+    cdef libc.stdint.int64_t v = signed_v & 0xFFFFFFFF
     self.write_var_int64(v)
 
   cpdef write_bigendian_int64(self, libc.stdint.int64_t signed_v):
@@ -156,7 +157,7 @@ cdef class ByteCountingOutputStream(OutputStream):
   cpdef write_var_int64(self, libc.stdint.int64_t signed_v):
     self.count += get_varint_size(signed_v)
 
-  cpdef write_var_int32(self, libc.stdint.int32_t signed_v):
+  cpdef write_var_int32(self, libc.stdint.int64_t signed_v):
     if signed_v < 0:
       self.count += 5
     else: