From 9f9b928d3d6e703d48cca562a2a56feab4830194 Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Thu, 22 Oct 2026 11:13:46 +0000 Subject: [PATCH 1/2] chore: bump org.apache.kafka:kafka-clients from 4.3.1 to 4.4.0 Bumps org.apache.kafka:kafka-clients from 4.3.1 to 4.4.0. --- updated-dependencies: - dependency-name: org.apache.kafka:kafka-clients dependency-version: 4.4.0 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] --- examples/powertools-examples-kafka/pom.xml | 2 +- powertools-kafka/pom.xml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/examples/powertools-examples-kafka/pom.xml b/examples/powertools-examples-kafka/pom.xml index e0282134a..c5024e23e 100644 --- a/examples/powertools-examples-kafka/pom.xml +++ b/examples/powertools-examples-kafka/pom.xml @@ -35,7 +35,7 @@ org.apache.kafka kafka-clients - 4.3.1 + 4.4.0 org.apache.avro diff --git a/powertools-kafka/pom.xml b/powertools-kafka/pom.xml index cef06b3e9..a3d634a64 100644 --- a/powertools-kafka/pom.xml +++ b/powertools-kafka/pom.xml @@ -34,7 +34,7 @@ - 4.3.1 + 4.4.0 1.12.2 4.35.0 1.4.1 From eed4de6aa7dbada9257c4e8c8d275c80e75edd7a Mon Sep 17 00:00:00 2001 From: Philipp Page Date: Fri, 23 Oct 2026 10:42:47 +0200 Subject: [PATCH 2/2] fix(kafka): replace internal ByteUtils with local varint decoding for kafka-clients 4.4.0 kafka-clients 4.4.0 moved org.apache.kafka.common.utils.ByteUtils to the internal package org.apache.kafka.common.utils.internals (KAFKA-20297, apache/kafka#21856). Decode Confluent Protobuf message indexes with a local zig-zag varint reader that matches ByteUtils.readVarint(ByteBuffer) exactly, so we no longer depend on Kafka internals. --- .../KafkaProtobufDeserializer.java | 33 +++++++++- .../KafkaProtobufDeserializerTest.java | 61 +++++++++++++++---- 2 files changed, 80 insertions(+), 14 deletions(-) diff --git a/powertools-kafka/src/main/java/software/amazon/lambda/powertools/kafka/serializers/KafkaProtobufDeserializer.java b/powertools-kafka/src/main/java/software/amazon/lambda/powertools/kafka/serializers/KafkaProtobufDeserializer.java index 1a8085a61..ba9e7f455 100644 --- a/powertools-kafka/src/main/java/software/amazon/lambda/powertools/kafka/serializers/KafkaProtobufDeserializer.java +++ b/powertools-kafka/src/main/java/software/amazon/lambda/powertools/kafka/serializers/KafkaProtobufDeserializer.java @@ -16,7 +16,6 @@ import java.lang.reflect.InvocationTargetException; import java.nio.ByteBuffer; -import org.apache.kafka.common.utils.ByteUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -38,6 +37,7 @@ public class KafkaProtobufDeserializer extends AbstractKafkaDeserializer { private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProtobufDeserializer.class); private static final String PROTOBUF_PARSER_METHOD = "parser"; + private static final int MAX_VARINT_BYTES = 5; @Override protected T deserializeObject(byte[] data, Class type, SchemaRegistryType schemaRegistryType) @@ -81,13 +81,13 @@ private T confluentDeserializer(byte[] data, Class type) LOGGER.debug("Using Confluent Deserializer"); Parser parser = (Parser) type.getMethod(PROTOBUF_PARSER_METHOD).invoke(null); ByteBuffer buffer = ByteBuffer.wrap(data); - int size = ByteUtils.readVarint(buffer); + int size = readVarint(buffer); // Only if the size is greater than zero, continue reading varInt. // https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#wire-format if (size > 0) { for (int i = 0; i < size; i++) { - ByteUtils.readVarint(buffer); + readVarint(buffer); } } Message message = parser.parseFrom(buffer); @@ -107,4 +107,31 @@ private T glueDeserializer(byte[] data, Class type) Message message = parser.parseFrom(codedInputStream); return type.cast(message); } + + /** + * Reads a zig-zag encoded variable-length integer as used by the Confluent Protobuf wire format for message + * indexes. Behaves identically to Kafka's {@code ByteUtils.readVarint(ByteBuffer)}, which moved to an internal + * package in kafka-clients 4.4.0 (KAFKA-20297) and is no longer part of the public API. + * + * @param buffer the buffer to read from + * @return the decoded integer + * @throws IllegalArgumentException if the varint does not terminate within 5 bytes + */ + static int readVarint(ByteBuffer buffer) { + int value = readUnsignedVarint(buffer); + return (value >>> 1) ^ -(value & 1); + } + + private static int readUnsignedVarint(ByteBuffer buffer) { + int result = 0; + for (int shift = 0; shift < MAX_VARINT_BYTES * 7; shift += 7) { + byte b = buffer.get(); + result |= (b & 0x7F) << shift; + if (b >= 0) { + return result; + } + } + throw new IllegalArgumentException("Varint is too long, the most significant bit in the 5th byte is set, " + + "converted value: " + Integer.toHexString(result)); + } } diff --git a/powertools-kafka/src/test/java/software/amazon/lambda/powertools/kafka/serializers/KafkaProtobufDeserializerTest.java b/powertools-kafka/src/test/java/software/amazon/lambda/powertools/kafka/serializers/KafkaProtobufDeserializerTest.java index 34a376947..1f614e0fe 100644 --- a/powertools-kafka/src/test/java/software/amazon/lambda/powertools/kafka/serializers/KafkaProtobufDeserializerTest.java +++ b/powertools-kafka/src/test/java/software/amazon/lambda/powertools/kafka/serializers/KafkaProtobufDeserializerTest.java @@ -17,9 +17,9 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.nio.BufferUnderflowException; import java.nio.ByteBuffer; -import org.apache.kafka.common.utils.ByteUtils; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -180,21 +180,60 @@ private byte[] createProtobufDataWithSimpleMessageIndexConfluent(TestProduct pro private byte[] createProtobufDataWithComplexMessageIndexConfluent(TestProduct product) throws IOException { ByteArrayOutputStream baos = new ByteArrayOutputStream(); - // Write complex message index array [1,0] using ByteUtils + // Write complex message index array [1,0] as zig-zag varints // https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#wire-format - ByteBuffer buffer = ByteBuffer.allocate(1024); - ByteUtils.writeVarint(2, buffer); // Array length - ByteUtils.writeVarint(1, buffer); // First index value - ByteUtils.writeVarint(0, buffer); // Second index value - - buffer.flip(); - byte[] indexData = new byte[buffer.remaining()]; - buffer.get(indexData); - baos.write(indexData); + baos.write(zigZagVarints(2, 1, 0)); // Array length, first index value, second index value // Write the protobuf data baos.write(product.toByteArray()); return baos.toByteArray(); } + + @Test + void shouldReadZigZagVarintsLikeKafkaByteUtils() throws IOException { + // Given + int[] values = { 0, 1, -1, 2, -2, 63, -64, 64, 300, -300, 16384, 2097151, 268435456, + Integer.MAX_VALUE, Integer.MIN_VALUE }; + ByteBuffer buffer = ByteBuffer.wrap(zigZagVarints(values)); + + // When/Then + for (int value : values) { + assertThat(KafkaProtobufDeserializer.readVarint(buffer)).isEqualTo(value); + } + assertThat(buffer.hasRemaining()).isFalse(); + } + + @Test + void shouldThrowWhenVarintExceedsFiveBytes() { + // Given + ByteBuffer buffer = ByteBuffer.wrap(new byte[] { (byte) 0x80, (byte) 0x80, (byte) 0x80, (byte) 0x80, + (byte) 0x80, 0x01 }); + + // When/Then + assertThatThrownBy(() -> KafkaProtobufDeserializer.readVarint(buffer)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Varint is too long"); + } + + @Test + void shouldThrowWhenVarintIsTruncated() { + // Given + ByteBuffer buffer = ByteBuffer.wrap(new byte[] { (byte) 0x80 }); + + // When/Then + assertThatThrownBy(() -> KafkaProtobufDeserializer.readVarint(buffer)) + .isInstanceOf(BufferUnderflowException.class); + } + + private byte[] zigZagVarints(int... values) throws IOException { + // Protobuf sint32 encoding is the same zig-zag varint encoding Confluent uses for message indexes + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + CodedOutputStream codedOutput = CodedOutputStream.newInstance(baos); + for (int value : values) { + codedOutput.writeSInt32NoTag(value); + } + codedOutput.flush(); + return baos.toByteArray(); + } }