[FLINK-40370][python] Frame bytes-backed decimals as a bytes field

[vbhanuchander-lang]

What is the purpose of the change

Bytes-backed Avro decimal logical types were encoded incompatibly with Flink's JVM serializer when a
GenericRecord crossed the JVM/Python boundary, silently corrupting the record.

FLINK-37192 replaced avro-python3 with avro>=1.12.0. That version routes a bytes-backed decimal
through BinaryEncoder.write_decimal_bytes and BinaryDecoder.read_decimal_from_bytes, both of
which size the payload with write_long / read_long. FlinkAvroEncoder and FlinkAvroDecoder
override the primitive reads and writes for JVM compatibility but not those two, and in this encoder
write_long is a fixed 8-byte long — so the length prefix was written as 8 bytes where the JVM
frames a bytes field with a 4-byte int.

For a record of amount=Decimal("12.34"), tail=7:

expected (JVM) : 0000000204d200000007
actual         : 000000000000000204d200000007

A JVM GenericDatumReader consumes the oversized prefix as the whole bytes field, so every following
field shifts: the decimal reads back empty and tail reads back as 2, leaving four bytes unread.
Nothing is raised, which is why this is silent corruption rather than a decode failure. Flink's own
reader is wrong in the same direction, so a pure-Python round trip looks correct and the defect only
appears across the boundary.

This does not affect standard Avro interoperability — avro, fastavro and ordinary Kafka Avro
payloads are unchanged. It is specific to Flink's internal JVM-compatible serializer.

Brief change log

• FlinkAvroEncoder.write_decimal_bytes frames the unscaled payload as a bytes field rather than
  letting avro size it with an 8-byte long
• FlinkAvroDecoder.read_decimal_from_bytes reads that 4-byte size symmetrically
• The payload itself is untouched: the same two's-complement big-endian unscaled value avro already
  produced, so only the length prefix moves

Verifying this change

This change added tests and can be verified as follows:

• pyflink/fn_execution/tests/test_avro_format.py, four tests: the exact encoded bytes for a
  decimal followed by an int; that the record reads back field by field with JVM framing; that the
  unscaled payload is unchanged for 0, ±12.34 and the ±1.28/-1.29 byte-boundary cases; and
  a round trip through FlinkAvroDatumWriter/FlinkAvroDatumReader
• Reverting the two overrides fails eight assertions across those tests
• Separately checked against avro 1.12.2 that the payload bytes stay identical to
  avro.io.BinaryEncoder.write_decimal_bytes across 23 values — positive, negative, zero and the
  ±128, ±256, ±32768 boundaries — so only the framing differs

A note on how that was run: the module imports only struct and avro, and I exercised the test
body with PyFlinkTestCase substituted by unittest.TestCase in a virtualenv with avro 1.12.2. I
did not build a full PyFlink environment locally, so the assertions are verified but the change has
not been through the project's own Python test harness on my machine. flake8 is clean under
flink-python/tox.ini (max-line-length 100).

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: yes — this is the JVM/Python boundary encoding for Avro generic records.
  It is a wire format used for in-flight data exchange rather than persisted state, and the
  previous encoding was already unreadable by the JVM, so the only case that changes behaviour is
  a decimal field that was silently corrupted before.
• The runtime per-record code paths (performance sensitive): yes — the decimal path only. The
  work is equivalent to what avro already did; one int.to_bytes replaces a manual byte loop.
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing,
  Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no — it corrects an encoding defect, so there
  is nothing to document.

[flinkbot]

CI report:

• 9840759 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build