aliehsaeedii commented on code in PR #21676:
URL: https://github.com/apache/kafka/pull/21676#discussion_r3559843453
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerRecord.java:
##########
@@ -81,6 +83,35 @@ public ProducerRecord(String topic, Integer partition, Long
timestamp, K key, V
this.value = value;
this.timestamp = timestamp;
this.headers = new RecordHeaders(headers);
+ this.rawSerializedHeaders = null;
+ }
+
+ /**
+ * Creates a record carrying pre-serialized header bytes. When this
constructor is used, the
+ * producer writes {@code rawSerializedHeaders} directly into the record
batch without
+ * deserializing or re-serializing individual headers. The {@code
rawSerializedHeaders} must
+ * use the standard Kafka header wire format: {@code
[count(varint)][header1][header2]...},
+ * or be empty (length 0) for zero headers.
+ *
+ * <p>This is intended for internal use (e.g., Kafka Streams changelog
writing) where headers
+ * are already available in serialized form and the deserialization
round-trip should be avoided.
+ */
+ public ProducerRecord(String topic, Integer partition, Long timestamp, K
key, V value, byte[] rawSerializedHeaders) {
Review Comment:
Oh, it needs a KIP. Also I don't like to add it for all users honestly. It's
sort of a foot-gun to be public and also we’d have to push hard to get that.
Based on my experience, this area is heavily guarded by the code owners. It
could take a year of effort and still result in no meaningful progress or
acceptance.
Suggested alternative (no public API, likely no KIP): keep the optimization
but smuggle the bytes through the *existing* constructor via an internal
carrier:
- Add an internal `Headers` implementation (e.g.
`common.header.internals.PreSerializedHeaders`) holding the raw bytes and
lazily materializing to `RecordHeaders` on any read/iteration.
- Streams passes it into the existing `ProducerRecord(...,
Iterable<Header>)` constructor — no new constructor.
- `ProducerRecord` special-cases it internally into a **package-private**
field; `KafkaProducer.doSend` (same package) reads a package-private getter and
routes to the raw append path.
- Because the carrier materializes on read, `headers()` stays correct and
interceptors keep working — `doSend` just falls back to the normal `Header[]`
path if anything already materialized it.
This matches the mechanism's blast radius to the (niche) benefit and stays
fully reversible.
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerRecord.java:
##########
@@ -185,6 +216,14 @@ public Integer partition() {
return partition;
}
+ /**
+ * @return Pre-serialized header bytes in Kafka wire format, or null if
headers were
+ * provided as {@link Headers} objects via the standard
constructors.
+ */
+ public byte[] rawSerializedHeaders() {
Review Comment:
This public getter is part of the same public-API expansion. Under the
internal-carrier approach this becomes a package-private getter read only by
`KafkaProducer.doSend` (same package), so it never enters the public surface.
##########
clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecord.java:
##########
@@ -222,6 +222,54 @@ 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)][header1][header2]...}, or be empty (length 0)
for zero headers.
+ */
+ public static int writeTo(DataOutputStream out,
+ int offsetDelta,
+ long timestampDelta,
+ ByteBuffer key,
+ ByteBuffer value,
+ byte[] rawSerializedHeaders) throws IOException {
+ int sizeInBytes = sizeOfBodyInBytes(offsetDelta, timestampDelta, key,
value, rawSerializedHeaders);
+ ByteUtils.writeVarint(sizeInBytes, out);
+
+ byte attributes = 0;
+ out.write(attributes);
+
+ ByteUtils.writeVarlong(timestampDelta, out);
+ ByteUtils.writeVarint(offsetDelta, out);
+
+ if (key == null) {
+ ByteUtils.writeVarint(-1, out);
+ } else {
+ int keySize = key.remaining();
+ ByteUtils.writeVarint(keySize, out);
+ Utils.writeTo(out, key, keySize);
+ }
+
+ if (value == null) {
+ ByteUtils.writeVarint(-1, out);
+ } else {
+ int valueSize = value.remaining();
+ ByteUtils.writeVarint(valueSize, out);
+ Utils.writeTo(out, value, valueSize);
+ }
+
+ if (rawSerializedHeaders == null)
+ throw new IllegalArgumentException("Raw serialized headers cannot
be null");
+
+ if (rawSerializedHeaders.length == 0) {
+ ByteUtils.writeVarint(0, out);
+ } else {
+ out.write(rawSerializedHeaders);
Review Comment:
The raw bytes are written verbatim with no validation. This is correct
**only** because the KIP-1271 stored-header format is byte-identical to the
on-wire `DefaultRecord` header format
(`[count(varint)][keyLen][key][valueLen|-1][value]...`). Worth an explicit
comment asserting that invariant. If the source of these bytes is ever a public
API (current constructor), malformed input silently corrupts the log; if it
comes from a trusted internal carrier instead, the risk is contained to
Streams-produced bytes.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java:
##########
@@ -146,6 +154,96 @@ 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);
+ final ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(
+ changelogPartition.topic(),
+ changelogPartition.partition(),
+ timestamp,
+ keyBytes,
+ value,
+ finalRawHeaders
+ );
+
+ collector.send(key, value, null, null, record);
+ }
+
+ /**
+ * Merge vector clock entries into raw serialized header bytes without
deserializing
+ * existing headers. The raw format is [count(varint)][entry1][entry2]...
or empty.
+ */
+ private byte[] mergeVectorClockIntoRawHeaders(final byte[] rawHeaders,
final Position position) {
Review Comment:
The vector-clock merge logic looks correct: it reads the leading count
varint, appends the two serialized VC entries, and rewrites `count + 2` without
deserializing existing headers. Two notes: (1) this allocates two
`ByteArrayOutputStream`/`DataOutputStream` per record when consistency is
enabled — fine, but on the hot path; consider sizing/reusing a buffer. (2) This
method is needed regardless of the public-vs-internal-carrier decision, so it
can stay as-is when switching to the internal carrier.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -401,6 +401,74 @@ private RecordAppendResult appendNewBatch(String topic,
return new RecordAppendResult(future, dq.size() > 1 || batch.isFull(),
true, batch.estimatedSizeInBytes());
}
+ public RecordAppendResult appendWithRawHeaders(String topic,
Review Comment:
`appendWithRawHeaders` / `tryAppendRawHeaders` / `appendNewBatchRawHeaders`
duplicate the existing `append`/`tryAppend`/`appendNewBatch` paths almost
verbatim, in the producer's hottest code. This is significant maintenance debt
(the two paths will drift). Consider parameterizing the existing methods over a
small "record body writer" abstraction (Header[] vs raw bytes) rather than
forking the whole append flow, so the partitioning/batching/allocation logic
stays single-sourced.
##########
clients/src/main/java/org/apache/kafka/clients/producer/ProducerRecord.java:
##########
@@ -81,6 +83,35 @@ public ProducerRecord(String topic, Integer partition, Long
timestamp, K key, V
this.value = value;
this.timestamp = timestamp;
this.headers = new RecordHeaders(headers);
+ this.rawSerializedHeaders = null;
+ }
+
+ /**
+ * Creates a record carrying pre-serialized header bytes. When this
constructor is used, the
+ * producer writes {@code rawSerializedHeaders} directly into the record
batch without
+ * deserializing or re-serializing individual headers. The {@code
rawSerializedHeaders} must
+ * use the standard Kafka header wire format: {@code
[count(varint)][header1][header2]...},
+ * or be empty (length 0) for zero headers.
+ *
+ * <p>This is intended for internal use (e.g., Kafka Streams changelog
writing) where headers
+ * are already available in serialized form and the deserialization
round-trip should be avoided.
+ */
+ public ProducerRecord(String topic, Integer partition, Long timestamp, K
key, V value, byte[] rawSerializedHeaders) {
+ if (topic == null)
+ throw new IllegalArgumentException("Topic cannot be null.");
+ if (timestamp != null && timestamp < 0)
+ throw new IllegalArgumentException(
+ String.format("Invalid timestamp: %d. Timestamp should
always be non-negative or null.", timestamp));
+ if (partition != null && partition < 0)
+ throw new IllegalArgumentException(
+ String.format("Invalid partition: %d. Partition number
should always be non-negative or null.", partition));
+ this.topic = topic;
+ this.partition = partition;
+ this.key = key;
+ this.value = value;
+ this.timestamp = timestamp;
+ this.headers = new RecordHeaders();
Review Comment:
With the raw path, `this.headers` is set to empty while the real headers
live in `rawSerializedHeaders`, so `record.headers()` no longer reflects the
record's actual headers. That's a silent trap: `ProducerInterceptor.onSend`
(which runs before `doSend`) sees empty headers, and any header edits an
interceptor makes are dropped. `equals`/`hashCode`/`toString` also become
half-consistent. The internal-carrier approach avoids this because it can
lazily materialize on read, keeping `headers()` accurate.
##########
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##########
@@ -1135,18 +1135,25 @@ 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;
+ byte[] rawHeaders = record.rawSerializedHeaders();
Review Comment:
Note the interceptor ordering: `this.interceptors.onSend(record)` runs in
`send(...)` before `doSend(...)`. Since the raw path leaves `record.headers()`
empty (see `ProducerRecord`), any configured `ProducerInterceptor` sees no
headers and cannot modify them. Streams' changelog producer typically has no
interceptors, but this is a behavioral divergence baked into a public type. A
lazily-materializing internal carrier would let `onSend` observe/modify the
real headers and only take the fast path when they were never materialized.
##########
streams/src/test/java/org/apache/kafka/test/InternalMockProcessorContext.java:
##########
@@ -477,6 +477,17 @@ 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) {
+ // In tests, delegate to the Headers-based overload via the regular
collector path
+ throw new UnsupportedOperationException("Raw header logChange not
supported in mock context");
Review Comment:
This will likely break existing tests.
`ChangeLoggingTimestampedKeyValueBytesStoreWithHeadersTest` and
`ChangeLoggingTimestampedWindowBytesStoreWithHeadersTest` drive these stores
through `InternalMockProcessorContext`, and the stores' non-null-value `put`
path now routes to this raw `logChange` overload — which throws here. Please
either implement this overload in the mock (delegate to the `Headers`-based
path by deserializing the raw bytes) or update those tests. Worth confirming
the streams test suite is green.
--
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]
