hutiefang76 opened a new issue, #12903:
URL: https://github.com/apache/seatunnel/issues/12903
### Search before asking
- [x] I searched issues and open/closed PRs for this converter and
ByteBuffer handling and found no matching fix.
### What happened
`AvroToRowConverter` converts an Avro `bytes` value using
`ByteBuffer.array()`. That returns the backing allocation rather than the
logical bytes between `position` and `limit`. A bounded or sliced heap buffer
therefore includes unrelated bytes; direct and read-only buffers throw
exceptions.
This also reproduces with a real Avro reader reusing a record: read a
64-byte value, then reuse the record for a 3-byte value. The buffer retains its
larger capacity, and the public converter returns 64 bytes instead of 3.
The scope is the public `converter(GenericRecord, SeaTunnelRowType)` and
`getReader()` APIs. The current Kafka/Pulsar `AvroDeserializationSchema` path
calls `read(null, ...)`, so it does not take the record-reuse path. I am not
claiming that normal connector payloads currently hit this failure.
### SeaTunnel Version
Current `dev`, commit `e2957d27afa29d4de4324cf642450757d865b902`
(3.0.1-SNAPSHOT).
### SeaTunnel Config
Direct converter API reproduction, with the real `BYTES` type
(`PrimitiveByteArrayType.INSTANCE`):
```java
SeaTunnelRowType type = new SeaTunnelRowType(
new String[] {"bytes"},
new SeaTunnelDataType<?>[] {PrimitiveByteArrayType.INSTANCE});
AvroSerializationSchema serializer = new AvroSerializationSchema(type);
AvroToRowConverter converter = new AvroToRowConverter(type);
DatumReader<GenericRecord> reader = converter.getReader();
byte[] first = serializer.serialize(new SeaTunnelRow(new Object[] {new
byte[64]}));
byte[] second = serializer.serialize(new SeaTunnelRow(new Object[] {new
byte[] {1, 2, 3}}));
GenericRecord oldRecord = reader.read(null,
DecoderFactory.get().binaryDecoder(first, null));
GenericRecord reused = reader.read(oldRecord,
DecoderFactory.get().binaryDecoder(second, null));
SeaTunnelRow row = converter.converter(reused, type);
Assertions.assertArrayEquals(new byte[] {1, 2, 3}, (byte[]) row.getField(0));
```
### Running Command
I added regressions to the existing `AvroConverterTest` and ran:
```sh
./mvnw -B -pl seatunnel-formats/seatunnel-format-avro -am
-Dtest=AvroConverterTest -Dsurefire.failIfNoSpecifiedTests=false package
```
### Error Exception
10 executed tests: 3 assertion failures (bounded heap, sliced heap, real
reader reuse), 2 errors (`UnsupportedOperationException` for a direct buffer
and `ReadOnlyBufferException` for a read-only buffer), 0 skips. The ordinary
heap, empty, null and normal serialization/deserialization controls passed, as
did the existing converter test.
The record-reuse assertion expected 3 bytes but received 64. No production
source was modified for these reproductions.
### Zeta or Flink or Spark Version
Not needed for the direct API reproduction.
### Java or Scala Version
JDK 17.
### Screenshots
Not applicable.
### Are you willing to submit PR?
Yes. I plan to read only the buffer's remaining region without changing its
position or limit, preserve the existing full-array fast path used by normal
connector reads, and keep the regressions in the existing test class. No
configuration or Avro wire-layout change is intended.
AI assistance was used for investigation and test drafting; the reported
failures and controls were actually executed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
