mjsax commented on code in PR #21762:
URL: https://github.com/apache/kafka/pull/21762#discussion_r2941808081
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/HeadersSerializer.java:
##########
@@ -48,50 +45,91 @@
*/
class HeadersSerializer {
+ static final class PreSerializedHeaders {
+ final int requiredBufferSizeForHeaders;
+ final byte[][] rawHeaderKeys;
+ final byte[][] rawHeaderValues;
+
+ PreSerializedHeaders(
+ final int requiredBufferSizeForHeaders,
+ final byte[][] rawHeaderKeys,
+ final byte[][] rawHeaderValues
+ ) {
+ this.requiredBufferSizeForHeaders = requiredBufferSizeForHeaders;
+ this.rawHeaderKeys = rawHeaderKeys;
+ this.rawHeaderValues = rawHeaderValues;
+ }
+ }
+
+ public static PreSerializedHeaders prepareSerialization(final Headers
headers) {
+ final Header[] headersArray = (headers == null) ? new Header[0] :
headers.toArray();
+
+ if (headersArray.length == 0) {
+ return new PreSerializedHeaders(0, null, null);
+ }
+
+ // we first compute the size for the buffer we need,
+ // so we can allocate the whole buffer at once later
+
+ // cache to avoid translating String header-keys to byte[] twice
+ final byte[][] serializerHeaderKeys = new byte[headersArray.length][];
+ final byte[][] serializedHeaderValues = new
byte[headersArray.length][];
+
+ // start with varint encoding of header count
+ int requiredBufferSizeForHeaders =
ByteUtils.sizeOfVarint(headersArray.length);
+
+ int i = 0;
+ for (final Header header : headersArray) {
+ serializerHeaderKeys[i] =
header.key().getBytes(StandardCharsets.UTF_8);
+ requiredBufferSizeForHeaders +=
ByteUtils.sizeOfVarint(serializerHeaderKeys[i].length) +
serializerHeaderKeys[i].length;
+
+ serializedHeaderValues[i] = header.value();
+ if (serializedHeaderValues[i] == null) {
+ ++requiredBufferSizeForHeaders;
+ } else {
+ requiredBufferSizeForHeaders +=
ByteUtils.sizeOfVarint(serializedHeaderValues[i].length) +
serializedHeaderValues[i].length;
+ }
+
+ ++i;
+ }
+
+ return new PreSerializedHeaders(requiredBufferSizeForHeaders,
serializerHeaderKeys, serializedHeaderValues);
+ }
+
/**
- * Serializes headers into a byte array using varint encoding per KIP-1271.
+ * Serializes headers into a ByteBuffer using varint encoding per KIP-1271.
* <p>
* The output format is [count][header1][header2]... without a size prefix.
* The size prefix is added by the outer serializer that uses this.
* <p>
* For null or empty headers, returns an empty byte array (0 bytes)
* instead of encoding headerCount=0 (1 byte).
*
- * @param headers the headers to serialize (can be null)
- * @return the serialized byte array (empty array if headers are null or
empty)
+ * @param preSerializedHeaders the preSerializedHeaders
+ * @param buffer the buffer to write the serialized header into (it's
expected that the buffer position is set correctly)
+ * @return the modified {@code buffer} containing the serializer headers
(empty array if headers are null or empty),\
Review Comment:
```suggestion
* @return the modified {@code buffer} containing the serializer headers
(empty array if headers are null or empty),
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
