This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch 4.2
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.2 by this push:
new 2d1f0c859f9 KAFKA-20769: ListDeserializer can silently deserialize a
corrupted entry when the input is truncated mid-entry (#22755)
2d1f0c859f9 is described below
commit 2d1f0c859f9c94a81f0b401bf63e4f5729f88a54
Author: Sepuri Sai Krishna <[email protected]>
AuthorDate: Fri Jul 24 11:01:20 2026 +0530
KAFKA-20769: ListDeserializer can silently deserialize a corrupted entry
when the input is truncated mid-entry (#22755)
ListDeserializer.deserialize is using DataInputStream.read(byte[]) which
may return "early" if the stream is shorter than expected (ie, cannot
fill the provided `byte[]` array) w/o error. This can lead to a
corrupted deserialization result.
This PR switches to DataInputStream.readFully, which throws EOFException
when the buffer cannot be filled.
Reviewers: Matthias J. Sax <[email protected]>
---
.../common/serialization/ListDeserializer.java | 8 ++++-
.../common/serialization/ListDeserializerTest.java | 35 ++++++++++++++++++++++
2 files changed, 42 insertions(+), 1 deletion(-)
diff --git
a/clients/src/main/java/org/apache/kafka/common/serialization/ListDeserializer.java
b/clients/src/main/java/org/apache/kafka/common/serialization/ListDeserializer.java
index da6ac3feff2..ce444550a22 100644
---
a/clients/src/main/java/org/apache/kafka/common/serialization/ListDeserializer.java
+++
b/clients/src/main/java/org/apache/kafka/common/serialization/ListDeserializer.java
@@ -28,6 +28,7 @@ import org.slf4j.LoggerFactory;
import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
+import java.io.EOFException;
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
@@ -189,7 +190,12 @@ public class ListDeserializer<Inner> implements
Deserializer<List<Inner>> {
continue;
}
byte[] payload = new byte[entrySize];
- if (dis.read(payload) == -1) {
+ try {
+ // Use readFully (not read) so a stream truncated
mid-entry is detected: read(byte[]) may
+ // return a partial count without reaching EOF, which
would silently leave the tail of the
+ // payload zero-filled and hand a corrupted entry to the
inner deserializer.
+ dis.readFully(payload);
+ } catch (EOFException e) {
log.error("Ran out of bytes in serialized list");
log.trace("Deserialized list so far: {}",
deserializedList); // avoid logging actual data above TRACE level since it may
contain sensitive information
throw new SerializationException("End of the stream was
reached prematurely");
diff --git
a/clients/src/test/java/org/apache/kafka/common/serialization/ListDeserializerTest.java
b/clients/src/test/java/org/apache/kafka/common/serialization/ListDeserializerTest.java
index 0127c79e69f..2f022456c63 100644
---
a/clients/src/test/java/org/apache/kafka/common/serialization/ListDeserializerTest.java
+++
b/clients/src/test/java/org/apache/kafka/common/serialization/ListDeserializerTest.java
@@ -25,6 +25,7 @@ import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -326,6 +327,40 @@ public class ListDeserializerTest {
);
}
+ @Test
+ public void shouldThrowOnEntryTruncatedMidStream() {
+ // entrySize (10) is within data.length (14) so it passes the bounds
check, but only 5 payload bytes
+ // actually follow, so deserialization must fail rather than return a
truncated entry.
+ final byte[] corruptedData = new byte[] {
+ (byte)
Serdes.ListSerde.SerializationStrategy.VARIABLE_SIZE.ordinal(),
+ (byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x01, // encodes
length == 1
+ (byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x0A, // encodes
entrySize == 10
+ (byte) 'h', (byte) 'e', (byte) 'l', (byte) 'l', (byte) 'o' // only
5 of the 10 declared bytes
+ };
+ final ListDeserializer<String> testDeserializer = new
ListDeserializer<>(ArrayList.class, new StringDeserializer());
+ final SerializationException exception = assertThrows(
+ SerializationException.class,
+ () -> testDeserializer.deserialize(null, corruptedData)
+ );
+ assertEquals(
+ "End of the stream was reached prematurely",
+ exception.getMessage()
+ );
+ }
+
+ @Test
+ public void shouldDeserializeListEndingInEmptyEntry() {
+ // A list whose last element serializes to zero bytes leaves the
stream at EOF when that final
+ // (empty) entry is read: the deserializer must handle this correctly
and should not fail.
+ final String topic = "topic";
+ final List<String> data = List.of("a", "");
+ final byte[] serializedData = new ListSerializer<>(new
StringSerializer()).serialize(topic, data);
+
+ final ListDeserializer<String> testDeserializer = new
ListDeserializer<>(ArrayList.class, new StringDeserializer());
+
+ assertEquals(data, testDeserializer.deserialize(topic,
serializedData));
+ }
+
@Test
public void shouldThrowOnNegativeNullEntryLength() {
final byte[] corruptedData = new byte[] {
