aliehsaeedii commented on code in PR #21676:
URL: https://github.com/apache/kafka/pull/21676#discussion_r3574614325
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java:
##########
@@ -146,6 +152,97 @@ public void logChange(final String storeName,
null);
}
+ @Override
+ public void logChange(final String storeName,
+ final Bytes key,
+ final byte[] value,
+ final long timestamp,
+ final byte[] rawSerializedHeaders,
+ final Position position) {
+ throwUnsupportedOperationExceptionIfStandby("logChange");
+
+ final TopicPartition changelogPartition =
stateManager().registeredChangelogPartitionFor(storeName);
+
+ byte[] finalRawHeaders = rawSerializedHeaders;
+ if (consistencyEnabled) {
+ finalRawHeaders =
mergeVectorClockIntoRawHeaders(rawSerializedHeaders, position);
+ }
+
+ final byte[] keyBytes =
BYTES_KEY_SERIALIZER.serialize(changelogPartition.topic(), null, key);
+ // Wrap the already-serialized header bytes in a lazy carrier so the
producer can write them
+ // verbatim, avoiding a deserialize/re-serialize round trip on the
changelog hot path.
+ final ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(
+ changelogPartition.topic(),
+ changelogPartition.partition(),
+ timestamp,
+ keyBytes,
+ value,
+ new PreSerializedHeaders(finalRawHeaders)
+ );
+
+ collector.send(key, value, null, null, record);
Review Comment:
This calls the 5-arg `send` directly, so it skips the `checkForException()`
that the `Headers`-based `logChange` runs via the 10-arg overload. A prior
async send failure now surfaces only at flush/commit, not at the next put.
Intended?
##########
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:
The whole optimization rests on these raw bytes being byte-identical to what
the `Header[]` path writes, but no test checks that. Add a test that writes the
same headers both ways and asserts the two record byte arrays are equal.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java:
##########
@@ -146,6 +152,97 @@ public void logChange(final String storeName,
null);
}
+ @Override
+ public void logChange(final String storeName,
+ final Bytes key,
+ final byte[] value,
+ final long timestamp,
+ final byte[] rawSerializedHeaders,
+ final Position position) {
+ throwUnsupportedOperationExceptionIfStandby("logChange");
+
+ final TopicPartition changelogPartition =
stateManager().registeredChangelogPartitionFor(storeName);
+
+ byte[] finalRawHeaders = rawSerializedHeaders;
+ if (consistencyEnabled) {
+ finalRawHeaders =
mergeVectorClockIntoRawHeaders(rawSerializedHeaders, position);
+ }
+
+ final byte[] keyBytes =
BYTES_KEY_SERIALIZER.serialize(changelogPartition.topic(), null, key);
+ // Wrap the already-serialized header bytes in a lazy carrier so the
producer can write them
+ // verbatim, avoiding a deserialize/re-serialize round trip on the
changelog hot path.
+ final ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(
+ changelogPartition.topic(),
+ changelogPartition.partition(),
+ timestamp,
+ keyBytes,
+ value,
+ new PreSerializedHeaders(finalRawHeaders)
+ );
+
+ collector.send(key, value, null, null, record);
+ }
+
+ /**
+ * Merge the two consistency vector-clock entries into raw serialized
header bytes without
+ * deserializing the existing headers. The raw format is
[count(varint)][entry1][entry2]...
+ * or empty.
+ * <p>
+ * This sits on the changelog hot path (once per record when consistency
is enabled), so it
+ * writes directly into a single exactly-sized {@link ByteBuffer} rather
than chaining
+ * {@code ByteArrayOutputStream}/{@code DataOutputStream} and intermediate
copies.
+ */
+ private byte[] mergeVectorClockIntoRawHeaders(final byte[] rawHeaders,
final Position position) {
Review Comment:
Do we have any dorect test here? The store tests go through
`InternalMockProcessorContext`, which materializes the bytes and calls the
`Headers` path instead, so this code never runs in tests. Could we add a test
with consistency enabled that checks the joined bytes deserialize to the
existing headers plus the two vector-clock entries?
##########
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##########
@@ -1169,18 +1170,32 @@ private Future<RecordMetadata> doSend(ProducerRecord<K,
V> record, Callback call
// take into account broker load, the amount of data produced to
each partition, etc.).
int partition = partition(record, serializedKey, serializedValue,
cluster);
- setReadOnly(record.headers());
- Header[] headers = record.headers().toArray();
-
- int serializedSize =
AbstractRecords.estimateSizeInBytesUpperBound(RecordBatch.CURRENT_MAGIC_VALUE,
- compression.type(), serializedKey, serializedValue,
headers);
- ensureValidRecordSize(serializedSize);
long timestamp = record.timestamp() == null ? nowMs :
record.timestamp();
- // Append the record to the accumulator. Note, that the actual
partition may be
- // calculated there and can be accessed via
appendCallbacks.topicPartition.
- RecordAccumulator.RecordAppendResult result =
accumulator.append(record.topic(), partition, timestamp, serializedKey,
- serializedValue, headers, appendCallbacks,
remainingWaitMs, nowMs, cluster);
+ final RecordAccumulator.RecordAppendResult result;
+ // Fast path: if the record carries headers that are already
serialized in the wire
+ // format and were never read/modified (e.g. Kafka Streams
changelog writes), hand the
+ // bytes straight to the accumulator without a
deserialize/re-serialize round trip. If
+ // anything materialized them (interceptor, serializer, headers()
read), rawIfUnmodified()
+ // returns null and we take the normal Header[] path.
+ byte[] rawHeaders = record.headers() instanceof
PreSerializedHeaders
Review Comment:
Neither this fast path nor the fallback (when `rawIfUnmodified()` returns
null because an interceptor or a `headers()` read materialized the headers) has
a producer-level test. Should we add tests for both?
##########
clients/src/main/java/org/apache/kafka/common/record/internal/MemoryRecordsBuilder.java:
##########
@@ -867,6 +915,24 @@ public boolean hasRoomFor(long timestamp, ByteBuffer key,
ByteBuffer value, Head
return this.writeLimit >= estimatedBytesWritten() + recordSize;
}
+ public boolean hasRoomFor(long timestamp, byte[] key, byte[] value, byte[]
rawSerializedHeaders) {
+ return hasRoomFor(timestamp, wrapNullable(key), wrapNullable(value),
rawSerializedHeaders);
+ }
+
+ public boolean hasRoomFor(long timestamp, ByteBuffer key, ByteBuffer
value, byte[] rawSerializedHeaders) {
Review Comment:
Unlike the `Header[]` sibling and the raw
`appendWithOffset`/`estimateSizeInBytesUpperBound` (which throw for magic <
V2), this always computes a V2 size, so it can report room for a record the raw
append would then reject. Not reachable today since the accumulator only uses
V2, but would be nice to keep the consistency.
--
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]
