[
https://issues.apache.org/jira/browse/KAFKA-20656?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18101176#comment-18101176
]
Yunseop Eom commented on KAFKA-20656:
-------------------------------------
PR opened: https://github.com/apache/kafka/pull/23053
Implemented KAFKA-20656 in Connect Struct. getBytes now returns the logical
remaining bytes of ByteBuffer values and supports direct buffers. Struct
equality and hashCode normalize direct BYTES fields so equivalent byte[] and
ByteBuffer representations compare consistently, while preserving the source
buffer's position and limit. No public API signature or wire format changed.
Validation: TDD RED reproduced failures for sliced buffers, direct buffers, and
byte[]/ByteBuffer equality. GREEN passed all new regression tests, the full
StructTest, and the full connect:api test suite. Spotless, Checkstyle,
SpotBugs, and git diff --check passed. Status: PR open and awaiting maintainer
review; not merged.
> 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)
