[
https://issues.apache.org/jira/browse/KAFKA-20656?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18101198#comment-18101198
]
Yunseop Eom commented on KAFKA-20656:
-------------------------------------
I checked the Kafka repository's in-tree Connect code and did not find a
connector that explicitly allocates a direct ByteBuffer, so this PR is not
based on a specific in-tree connector incident.
The motivation is API-contract correctness rather than claiming that direct
buffers are common. ConnectSchema explicitly accepts both byte[] and ByteBuffer
for BYTES and notes that ByteBuffer is the recommended representation. The
current Struct.getBytes implementation calls ByteBuffer.array(), which returns
the backing array rather than the logical remaining bytes and throws for direct
buffers. External connectors are also free to provide either supported
representation.
The regression tests therefore cover the contract boundary (sliced buffers,
direct buffers, and equivalent byte[]/ByteBuffer Struct values). If maintainers
prefer to limit the scope to the remaining-byte and equality issues, I can
revise the direct-buffer test and implementation accordingly, but removing it
would leave an accepted BYTES representation unsupported by Struct.getBytes.
> Struct BYTES handling does not normalize ByteBuffer values
> ----------------------------------------------------------
>
> Key: KAFKA-20656
> URL: https://issues.apache.org/jira/browse/KAFKA-20656
> Project: Kafka
> Issue Type: Bug
> Components: connect
> Reporter: Ruiqi Dong
> Priority: Major
>
> *Summary*
> Kafka Connect `Schema.Type.BYTES` explicitly accepts both `byte[]` and
> `ByteBuffer`. `ConnectSchema` even recommends `ByteBuffer` because plain
> arrays do not implement content-based `equals()` / `hashCode()`. `Struct`
> does not consistently honor that contract:
> - `getBytes(...)` calls `ByteBuffer.array()`, which returns the whole backing
> array instead of the logical remaining bytes.
> - `getBytes(...)` throws `UnsupportedOperationException` for direct buffers.
> - `equals(...)` and `hashCode()` compare raw stored objects, so equivalent
> `byte[]` and `ByteBuffer` BYTES values are not equal.
>
> *Affected code*
> File: connect/api/src/main/java/org/apache/kafka/connect/data/Struct.java
> {code:java}
> public byte[] getBytes(String fieldName) {
> Object bytes = getCheckType(fieldName, Schema.Type.BYTES);
> if (bytes instanceof ByteBuffer)
> return ((ByteBuffer) bytes).array();
> return (byte[]) bytes;
> }
> @Override
> public boolean equals(Object o) {
> if (this == o) return true;
> if (o == null || getClass() != o.getClass()) return false;
> Struct struct = (Struct) o;
> return Objects.equals(schema, struct.schema) &&
> Arrays.deepEquals(values, struct.values);
> } {code}
> `Values.convertToBytes(...)` already uses `Utils.toArray((ByteBuffer)
> value)`, which preserves the buffer's remaining bytes and supports direct
> buffers.
>
> *Reproducer*
> Add these tests to
> connect/api/src/test/java/org/apache/kafka/connect/data/StructTest.java:
> {code:java}
> @Test
> public void testGetBytesPreservesByteBufferRemainingBytes() {
> ByteBuffer bytes = ByteBuffer.wrap(new byte[] {1, 2, 3, 4});
> bytes.position(1);
> bytes.limit(3);
> Struct struct = new Struct(FLAT_STRUCT_SCHEMA)
> .put("int8", (byte) 12)
> .put("int16", (short) 12)
> .put("int32", 12)
> .put("int64", (long) 12)
> .put("float32", 12.f)
> .put("float64", 12.)
> .put("boolean", true)
> .put("string", "foobar")
> .put("bytes", bytes);
> assertArrayEquals(new byte[] {2, 3}, struct.getBytes("bytes"));
> }
> @Test
> public void testGetBytesSupportsDirectByteBuffer() {
> ByteBuffer bytes = ByteBuffer.allocateDirect(2);
> bytes.put((byte) 1);
> bytes.put((byte) 2);
> bytes.flip();
> Struct struct = new Struct(FLAT_STRUCT_SCHEMA)
> .put("int8", (byte) 12)
> .put("int16", (short) 12)
> .put("int32", 12)
> .put("int64", (long) 12)
> .put("float32", 12.f)
> .put("float64", 12.)
> .put("boolean", true)
> .put("string", "foobar")
> .put("bytes", bytes);
> assertArrayEquals(new byte[] {1, 2}, struct.getBytes("bytes"));
> }
> @Test
> public void testEqualsAndHashCodeWithEquivalentByteArrayAndByteBufferValues()
> {
> Struct byteArrayStruct = new Struct(FLAT_STRUCT_SCHEMA)
> .put("int8", (byte) 12)
> .put("int16", (short) 12)
> .put("int32", 12)
> .put("int64", (long) 12)
> .put("float32", 12.f)
> .put("float64", 12.)
> .put("boolean", true)
> .put("string", "foobar")
> .put("bytes", "foobar".getBytes());
> Struct byteBufferStruct = new Struct(FLAT_STRUCT_SCHEMA)
> .put("int8", (byte) 12)
> .put("int16", (short) 12)
> .put("int32", 12)
> .put("int64", (long) 12)
> .put("float32", 12.f)
> .put("float64", 12.)
> .put("boolean", true)
> .put("string", "foobar")
> .put("bytes", ByteBuffer.wrap("foobar".getBytes()));
> assertEquals(byteArrayStruct, byteBufferStruct);
> assertEquals(byteArrayStruct.hashCode(), byteBufferStruct.hashCode());
> } {code}
> Run:
> {code:java}
> ./gradlew -q :connect:api:test \
> --tests
> org.apache.kafka.connect.data.StructTest.testGetBytesPreservesByteBufferRemainingBytes
> \
> --tests
> org.apache.kafka.connect.data.StructTest.testGetBytesSupportsDirectByteBuffer
> \
> --tests
> org.apache.kafka.connect.data.StructTest.testEqualsAndHashCodeWithEquivalentByteArrayAndByteBufferValues{code}
> Observed behavior:
> All three tests fail
> {code:java}
> array lengths differ, expected: <2> but was: <4>
> java.lang.UnsupportedOperationException
> expected Struct{...,bytes=[B@...} but was
> Struct{...,bytes=java.nio.HeapByteBuffer[pos=0 lim=6 cap=6]} {code}
> Expected behavior:
> `Struct.getBytes(...)` should return the logical BYTES value represented by
> the buffer's current remaining bytes and should support direct buffers.
> Equivalent BYTES values should compare consistently regardless of whether the
> value was supplied as `byte[]` or `ByteBuffer`.
>
> `ByteBuffer` is a documented valid representation for Connect BYTES values.
> Using `ByteBuffer.array()` exposes storage details instead of the logical
> data and rejects valid direct buffers. This makes a public typed getter
> unreliable for one of the two supported BYTES representations.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
