mjsax commented on code in PR #13533:
URL: https://github.com/apache/kafka/pull/13533#discussion_r1169327746
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/ChangedSerializer.java:
##########
@@ -113,9 +113,9 @@ public byte[] serialize(final String topic, final Headers
headers, final Change<
throw new StreamsException("Both old and new values are not
null (" + data.oldValue
+ " : " + data.newValue + ") in ChangeSerializer,
which is not allowed unless upgrading.");
} else {
- final int capacity = UINT32_SIZE + newDataLength +
oldDataLength + NEW_OLD_FLAG_SIZE;
+ final int capacity = MAX_VARINT_LENGTH + newDataLength +
oldDataLength + NEW_OLD_FLAG_SIZE;
buf = ByteBuffer.allocate(capacity);
- ByteUtils.writeUnsignedInt(buf, newDataLength);
+ ByteUtils.writeVarint(newDataLength, buf);
buf.put(newData).put(oldData).put((byte) 2);
Review Comment:
I would like to save second bytebuffer allocation -- we already have a
`ByteBuffer` at hand, so copying twice seems to be wasteful?
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
