muralibasani commented on code in PR #21676:
URL: https://github.com/apache/kafka/pull/21676#discussion_r3581487582
##########
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:
Good one. This was not intentional. I have restored that check and added
checkForException() at the top of the 5 arg. Looks aligned now based on your
comment..
--
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]
