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
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();
+ }
}