You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2026/08/20 17:37:15 UTC
(kafka) branch 4.2 updated (ccacf8ab551 -> 36689f3db1f)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a change to branch 4.2
in repository https://gitbox.apache.org/repos/asf/kafka.git
from ccacf8ab551 KAFKA-20870: Add test for the owned-task report for a single-role assignment change (#23002)
new 5a905ca7f86 MINOR: improve ProcessorMetadata (#23156)
new 4f16409650e MINOR: improve ChangedDeserializer (#23157)
new 96626e3e32a MINOR: improve ProcessorRecordContext (#23158)
new 36689f3db1f MINOR: improve CombinedKeySchema (#23159)
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../kstream/internals/ChangedDeserializer.java | 7 ++++
.../foreignkeyjoin/CombinedKeySchema.java | 4 +++
.../processor/internals/ProcessorMetadata.java | 8 +++++
.../internals/ProcessorRecordContext.java | 4 +++
.../kstream/internals/ChangedSerdeTest.java | 41 ++++++++++++++++++++++
.../foreignkeyjoin/CombinedKeySchemaTest.java | 22 ++++++++++++
.../processor/internals/ProcessorMetadataTest.java | 39 ++++++++++++++++++++
.../internals/ProcessorRecordContextTest.java | 39 ++++++++++++++++++++
8 files changed, 164 insertions(+)
(kafka) 03/04: MINOR: improve ProcessorRecordContext (#23158)
Posted by mj...@apache.org.
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
commit 96626e3e32a6ac4e4967533455b79d884452388e
Author: Lucy Liu <lu...@confluent.io>
AuthorDate: Thu Aug 20 12:10:02 2026 -0500
MINOR: improve ProcessorRecordContext (#23158)
Reviewers: Matthias J. Sax <ma...@confluent.io>
---
.../internals/ProcessorRecordContext.java | 4 +++
.../internals/ProcessorRecordContextTest.java | 39 ++++++++++++++++++++++
2 files changed, 43 insertions(+)
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContext.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContext.java
index 8198645eb65..72ffcdd3a16 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContext.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContext.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.streams.processor.internals;
+import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeader;
@@ -195,6 +196,9 @@ public class ProcessorRecordContext implements RecordContext, RecordMetadata {
if (headerCount == -1) { // keep for backward compatibility
headers = new RecordHeaders();
} else {
+ if (headerCount > buffer.remaining() / (2 * Integer.BYTES)) {
+ throw new SerializationException();
+ }
final Header[] headerArr = new Header[headerCount];
for (int i = 0; i < headerCount; i++) {
final byte[] keyBytes = requireNonNull(getNullableSizePrefixedArray(buffer));
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContextTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContextTest.java
index 1324e706268..62aeac5dc1b 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContextTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContextTest.java
@@ -16,11 +16,15 @@
*/
package org.apache.kafka.streams.processor.internals;
+import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.junit.jupiter.api.Test;
+import java.nio.ByteBuffer;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -110,4 +114,39 @@ public class ProcessorRecordContextTest {
assertEquals(MIN_SIZE + 10L, context.residentMemorySizeEstimate());
}
+
+ @Test
+ public void shouldRejectHeaderCountLargerThanRemainingBuffer() {
+ final Headers headers = new RecordHeaders();
+ headers.add("header-key", "header-value".getBytes(UTF_8));
+ final ProcessorRecordContext context = new ProcessorRecordContext(
+ 42L, 73L, 0, "topic", headers);
+
+ final byte[] serialized = context.serialize();
+ final ByteBuffer buffer = ByteBuffer.wrap(serialized);
+
+ // Locate headerCount's position within the real serialized bytes
+ // timestamp(8) + offset(8) + topicLen(4) + "topic"(5) + partition(4)
+ final int headerCountOffset = 8 + 8 + 4 + "topic".getBytes(UTF_8).length + 4;
+ final int bytesAfterHeaderCount = serialized.length - (headerCountOffset + 4);
+ final int maxPlausibleHeaderCount = bytesAfterHeaderCount / (2 * Integer.BYTES);
+
+ // Overwrite the real headerCount with a little more that it could support
+ buffer.putInt(headerCountOffset, maxPlausibleHeaderCount + 1);
+
+ assertThrows(SerializationException.class, () -> ProcessorRecordContext.deserialize(buffer));
+ }
+
+ @Test
+ public void shouldDeserializeValidHeaderCountWithoutRejecting() {
+ final Headers headers = new RecordHeaders();
+ headers.add("header-key", "header-value".getBytes(UTF_8));
+ final ProcessorRecordContext context = new ProcessorRecordContext(
+ 42L, 73L, 0, "topic", headers);
+
+ final ProcessorRecordContext roundTripped =
+ ProcessorRecordContext.deserialize(ByteBuffer.wrap(context.serialize()));
+
+ assertEquals(context, roundTripped);
+ }
}
(kafka) 04/04: MINOR: improve CombinedKeySchema (#23159)
Posted by mj...@apache.org.
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
commit 36689f3db1f7e6c64d8735736b38b747c6555409
Author: Lucy Liu <lu...@confluent.io>
AuthorDate: Thu Aug 20 12:11:14 2026 -0500
MINOR: improve CombinedKeySchema (#23159)
Reviewers: Matthias J. Sax <ma...@confluent.io>
---
.../foreignkeyjoin/CombinedKeySchema.java | 4 ++++
.../foreignkeyjoin/CombinedKeySchemaTest.java | 22 ++++++++++++++++++++++
2 files changed, 26 insertions(+)
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKeySchema.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKeySchema.java
index 176205c7248..f4ba07180ad 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKeySchema.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKeySchema.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.streams.kstream.internals.foreignkeyjoin;
+import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serializer;
@@ -84,6 +85,9 @@ public class CombinedKeySchema<KRight, KLeft> {
final byte[] dataArray = data.get();
final ByteBuffer dataBuffer = ByteBuffer.wrap(dataArray);
final int foreignKeyLength = dataBuffer.getInt();
+ if (foreignKeyLength > dataBuffer.remaining()) {
+ throw new SerializationException();
+ }
final byte[] foreignKeyRaw = new byte[foreignKeyLength];
dataBuffer.get(foreignKeyRaw, 0, foreignKeyLength);
final KRight foreignKey = foreignKeyDeserializer.deserialize(foreignKeySerdeTopic, foreignKeyRaw);
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKeySchemaTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKeySchemaTest.java
index 3142f65febf..0cf54a5ed84 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKeySchemaTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKeySchemaTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.streams.kstream.internals.foreignkeyjoin;
+import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
@@ -27,6 +28,8 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class CombinedKeySchemaTest {
+ private static final String FK_TOPIC = "fkTopic";
+ private static final String PK_TOPIC = "pkTopic";
@Test
public void nonNullPrimaryKeySerdeTest() {
@@ -88,4 +91,23 @@ public class CombinedKeySchemaTest {
final String foreignKey = null;
assertThrows(NullPointerException.class, () -> cks.prefixBytes(foreignKey));
}
+
+ @Test
+ public void shouldThrowWhenForeignKeyLengthExceedsRemainingBytes() {
+ final CombinedKeySchema<String, Integer> cks = new CombinedKeySchema<>(
+ () -> FK_TOPIC, Serdes.String(),
+ () -> PK_TOPIC, Serdes.Integer()
+ );
+
+ final byte[] foreignKeyRaw = Serdes.String().serializer().serialize(FK_TOPIC, "foreignKey");
+ final byte[] primaryKeyRaw = Serdes.Integer().serializer().serialize(PK_TOPIC, 1);
+ final int fakeForeignKeyLength = foreignKeyRaw.length + primaryKeyRaw.length + 1; // one past what's actually there
+
+ final ByteBuffer buf = ByteBuffer.allocate(Integer.BYTES + foreignKeyRaw.length + primaryKeyRaw.length);
+ buf.putInt(fakeForeignKeyLength);
+ buf.put(foreignKeyRaw).put(primaryKeyRaw);
+ final Bytes corrupted = Bytes.wrap(buf.array());
+
+ assertThrows(SerializationException.class, () -> cks.fromBytes(corrupted));
+ }
}
(kafka) 02/04: MINOR: improve ChangedDeserializer (#23157)
Posted by mj...@apache.org.
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
commit 4f16409650e8159ab70c17f415a47b6440294e9d
Author: Lucy Liu <lu...@confluent.io>
AuthorDate: Thu Aug 20 12:09:47 2026 -0500
MINOR: improve ChangedDeserializer (#23157)
Reviewers: Matthias J. Sax <ma...@confluent.io>
---
.../kstream/internals/ChangedDeserializer.java | 7 ++++
.../kstream/internals/ChangedSerdeTest.java | 41 ++++++++++++++++++++++
2 files changed, 48 insertions(+)
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/ChangedDeserializer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/ChangedDeserializer.java
index 3ae46843289..7fc23fdb9cd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/ChangedDeserializer.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/ChangedDeserializer.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.streams.kstream.internals;
+import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.utils.ByteUtils;
@@ -81,6 +82,9 @@ public class ChangedDeserializer<T> implements Deserializer<Change<T>>, Wrapping
}
case (byte) 2: {
final int newDataLength = ByteUtils.readVarint(buffer);
+ if (newDataLength > buffer.remaining()) {
+ throw new SerializationException();
+ }
newData = new byte[newDataLength];
buffer.get(newData);
@@ -108,6 +112,9 @@ public class ChangedDeserializer<T> implements Deserializer<Change<T>>, Wrapping
}
case (byte) 5: {
final int newDataLength = ByteUtils.readVarint(buffer);
+ if (newDataLength > buffer.remaining()) {
+ throw new SerializationException();
+ }
newData = new byte[newDataLength];
buffer.get(newData);
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/ChangedSerdeTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/ChangedSerdeTest.java
index 4d1b2b0b212..c2bab5099dd 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/ChangedSerdeTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/ChangedSerdeTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.streams.kstream.internals;
+import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.Serializer;
@@ -176,4 +177,44 @@ public class ChangedSerdeTest {
final Change<String> deserialized = CHANGED_STRING_DESERIALIZER.deserialize(TOPIC, serialized);
assertThat(deserialized, is(data));
}
+
+ @Test
+ public void shouldThrowOnNewDataLengthLargerThanRemainingBytesForFlag2() {
+ final byte[] newData = STRING_SERIALIZER.serialize(TOPIC, nonNullNewValue);
+ final byte[] oldData = STRING_SERIALIZER.serialize(TOPIC, nonNullOldValue);
+ final int fakeNewDataLength = newData.length + oldData.length + ENCODING_FLAG_SIZE + 1; // one past what the buffer can actually back
+
+ final int capacity = MAX_VARINT_LENGTH + newData.length + oldData.length + ENCODING_FLAG_SIZE;
+ final ByteBuffer buf = ByteBuffer.allocate(capacity);
+ ByteUtils.writeVarint(fakeNewDataLength, buf);
+ buf.put(newData).put(oldData).put((byte) 2);
+ final byte[] serialized = new byte[buf.position()];
+ buf.position(0);
+ buf.get(serialized);
+
+ assertThrows(
+ SerializationException.class,
+ () -> CHANGED_STRING_DESERIALIZER.deserialize(TOPIC, serialized)
+ );
+ }
+
+ @Test
+ public void shouldThrowOnNewDataLengthLargerThanRemainingBytesForFlag5() {
+ final byte[] newData = STRING_SERIALIZER.serialize(TOPIC, nonNullNewValue);
+ final byte[] oldData = STRING_SERIALIZER.serialize(TOPIC, nonNullOldValue);
+ final int fakeNewDataLength = newData.length + oldData.length + IS_LATEST_FLAG_SIZE + ENCODING_FLAG_SIZE + 1;
+
+ final int capacity = MAX_VARINT_LENGTH + newData.length + oldData.length + IS_LATEST_FLAG_SIZE + ENCODING_FLAG_SIZE;
+ final ByteBuffer buf = ByteBuffer.allocate(capacity);
+ ByteUtils.writeVarint(fakeNewDataLength, buf);
+ buf.put(newData).put(oldData).put((byte) 1).put((byte) 5);
+ final byte[] serialized = new byte[buf.position()];
+ buf.position(0);
+ buf.get(serialized);
+
+ assertThrows(
+ SerializationException.class,
+ () -> CHANGED_STRING_DESERIALIZER.deserialize(TOPIC, serialized)
+ );
+ }
}
(kafka) 01/04: MINOR: improve ProcessorMetadata (#23156)
Posted by mj...@apache.org.
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
commit 5a905ca7f86ae3aa7d61672cc182564c8eec7eb6
Author: Lucy Liu <lu...@confluent.io>
AuthorDate: Thu Aug 20 12:09:34 2026 -0500
MINOR: improve ProcessorMetadata (#23156)
Reviewers: Matthias J. Sax <ma...@confluent.io>
---
.../processor/internals/ProcessorMetadata.java | 8 +++++
.../processor/internals/ProcessorMetadataTest.java | 39 ++++++++++++++++++++++
2 files changed, 47 insertions(+)
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorMetadata.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorMetadata.java
index 943beb64e01..792eb592ebc 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorMetadata.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorMetadata.java
@@ -16,6 +16,8 @@
*/
package org.apache.kafka.streams.processor.internals;
+import org.apache.kafka.common.errors.SerializationException;
+
import java.io.ByteArrayOutputStream;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
@@ -52,9 +54,15 @@ public class ProcessorMetadata {
final ByteBuffer buffer = ByteBuffer.wrap(metaDataBytes);
final int entrySize = buffer.getInt();
+ if (entrySize > buffer.remaining() / (Integer.BYTES + Long.BYTES)) {
+ throw new SerializationException();
+ }
final Map<String, Long> metadata = new HashMap<>(entrySize);
for (int i = 0; i < entrySize; i++) {
final int keySize = buffer.getInt();
+ if (keySize > buffer.remaining()) {
+ throw new SerializationException();
+ }
final byte[] keyBytes = new byte[keySize];
buffer.get(keyBytes);
final Long value = buffer.getLong();
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorMetadataTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorMetadataTest.java
index edd276e850b..8d3f198c7a8 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorMetadataTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorMetadataTest.java
@@ -16,8 +16,12 @@
*/
package org.apache.kafka.streams.processor.internals;
+import org.apache.kafka.common.errors.SerializationException;
+
import org.junit.jupiter.api.Test;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
@@ -26,6 +30,7 @@ import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class ProcessorMetadataTest {
@@ -161,4 +166,38 @@ public class ProcessorMetadataTest {
assertEquals(metadata1, metadata2);
assertEquals(metadata1.hashCode(), metadata2.hashCode());
}
+
+ @Test
+ public void shouldThrowWhenKeySizeExceedsRemainingBytes() {
+ final byte[] keyBytes = "key1".getBytes(StandardCharsets.UTF_8);
+ final int fakeKeySize = keyBytes.length + Long.BYTES + 1; // one past what's actually left for this entry
+
+ final ByteBuffer buf = ByteBuffer.allocate(Integer.BYTES + Integer.BYTES + keyBytes.length + Long.BYTES);
+ buf.putInt(1); // entrySize = 1, legitimate
+ buf.putInt(fakeKeySize); // keySize is too large
+ buf.put(keyBytes);
+ buf.putLong(1L);
+ final byte[] serialized = new byte[buf.position()];
+ buf.position(0);
+ buf.get(serialized);
+
+ assertThrows(SerializationException.class, () -> ProcessorMetadata.deserialize(serialized));
+ }
+
+ @Test
+ public void shouldThrowWhenEntrySizeExceedsRemainingBytes() {
+ final byte[] keyBytes = "k".getBytes(StandardCharsets.UTF_8);
+ final int fakeEntrySize = 2; // claims 2 entries, but only enough real bytes exist for 1
+
+ final ByteBuffer buf = ByteBuffer.allocate(Integer.BYTES + Integer.BYTES + keyBytes.length + Long.BYTES);
+ buf.putInt(fakeEntrySize); // entrySize too large
+ buf.putInt(keyBytes.length); // keySize for the one real entry that follows
+ buf.put(keyBytes);
+ buf.putLong(1L);
+ final byte[] serialized = new byte[buf.position()];
+ buf.position(0);
+ buf.get(serialized);
+
+ assertThrows(SerializationException.class, () -> ProcessorMetadata.deserialize(serialized));
+ }
}