muralibasani commented on code in PR #21676:
URL: https://github.com/apache/kafka/pull/21676#discussion_r3581501466
##########
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:
Added a test. Now it enables consistency and covers the gap to drive that
ProcessorContextImpl raw path.
--
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]
