muralibasani commented on code in PR #21676:
URL: https://github.com/apache/kafka/pull/21676#discussion_r3564506948
##########
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:
Thanks for confirming that logic. Addressed both the notes.
--
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]
