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:23:56 UTC
(kafka) branch 4.3 updated (9a19890c91f -> 5484148fce7)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a change to branch 4.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
from 9a19890c91f KAFKA-20845: Don't flush empty batch when appending large records in group coordinator (#22969)
new e469c986f29 MINOR: improve ProcessorMetadata (#23156)
new 9eaff7e455a MINOR: improve ChangedDeserializer (#23157)
new e1a1ec83890 MINOR: improve ProcessorRecordContext (#23158)
new 5484148fce7 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 | 20 +++++++++++
.../processor/internals/ProcessorMetadataTest.java | 39 ++++++++++++++++++++
.../internals/ProcessorRecordContextTest.java | 39 ++++++++++++++++++++
8 files changed, 162 insertions(+)
(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.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 9eaff7e455a914ad88ca5f7cbb03f081bf1e41ef
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 51358eb5867..6c42a7cdbba 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 426b4b4ed5c..d6d533a73f2 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.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.serialization.Deserializer;
@@ -216,4 +217,44 @@ public class ChangedSerdeTest {
verify(mockDeserializer).deserialize(TOPIC, HEADERS, value.getBytes());
verify(mockDeserializer, never()).deserialize(TOPIC, value.getBytes());
}
+
+ @Test
+ public void shouldThrowOnNewDataLengthLargerThanRemainingBytesForFlag2() {
+ final byte[] newData = STRING_SERIALIZER.serialize(TOPIC, HEADERS, nonNullNewValue);
+ final byte[] oldData = STRING_SERIALIZER.serialize(TOPIC, HEADERS, 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, HEADERS, serialized)
+ );
+ }
+
+ @Test
+ public void shouldThrowOnNewDataLengthLargerThanRemainingBytesForFlag5() {
+ final byte[] newData = STRING_SERIALIZER.serialize(TOPIC, HEADERS, nonNullNewValue);
+ final byte[] oldData = STRING_SERIALIZER.serialize(TOPIC, HEADERS, 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, HEADERS, 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.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit e469c986f29f6c2afc0d49efa0d5998174a37e5d
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));
+ }
}
(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.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit e1a1ec8389011e0b8ea9d0ab728dcf6d41d368c8
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.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 5484148fce70a485906561289ef3ce5335b7a0ed
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>
---
.../internals/foreignkeyjoin/CombinedKeySchema.java | 4 ++++
.../foreignkeyjoin/CombinedKeySchemaTest.java | 20 ++++++++++++++++++++
2 files changed, 24 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 4d40856a990..f3802c38084 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.header.Headers;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
@@ -91,6 +92,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, headers, 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 22bb83e5d68..9ed6c57a335 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.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.serialization.Deserializer;
@@ -184,4 +185,23 @@ public class CombinedKeySchemaTest {
verify(mockDeserializer).deserialize(PK_TOPIC, HEADERS, primaryKeyRaw);
verify(mockDeserializer, never()).deserialize(PK_TOPIC, primaryKeyRaw);
}
+
+ @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, HEADERS));
+ }
}