muralibasani commented on code in PR #21676:
URL: https://github.com/apache/kafka/pull/21676#discussion_r3581506698
##########
clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecord.java:
##########
@@ -222,6 +222,63 @@ public static int writeTo(DataOutputStream out,
return ByteUtils.sizeOfVarint(sizeInBytes) + sizeInBytes;
}
+ /**
+ * Write the record to {@code out} using pre-serialized header bytes,
bypassing per-header
+ * iteration. The {@code rawSerializedHeaders} must use the standard Kafka
header wire format:
+ * {@code
[count(varint)][keyLen(varint)][key][valueLen(varint)|-1][value]...}, or be
empty
+ * (length 0) for zero headers.
+ *
+ * <p><b>Format invariant:</b> the bytes are written verbatim with no
validation, so this is
+ * correct <em>only</em> because the header section of the on-wire {@link
DefaultRecord} format
+ * is byte-identical to the format the callers pre-serialize (the KIP-1271
stored-header format
+ * and {@code Headers} serialization both use this exact layout). The
empty-headers case is
+ * still normalized to a {@code 0} count varint here. Callers must
therefore supply bytes from a
+ * trusted internal producer (e.g. Kafka Streams changelog writes);
feeding malformed bytes
+ * would silently corrupt the record.
+ */
+ public static int writeTo(DataOutputStream out,
+ int offsetDelta,
+ long timestampDelta,
+ ByteBuffer key,
+ ByteBuffer value,
+ byte[] rawSerializedHeaders) throws IOException {
Review Comment:
Agree, added a test for this one.
--
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]
