aliehsaeedii commented on code in PR #22961:
URL: https://github.com/apache/kafka/pull/22961#discussion_r3667522259
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RecordConverters.java:
##########
@@ -105,6 +105,40 @@ public static RecordConverter
rawValueToSessionHeadersValue() {
return RAW_TO_SESSION_WITH_HEADERS_INSTANCE;
}
+ private static final RecordConverter RAW_LIST_TO_HEADERS_LIST_INSTANCE =
record -> {
+ // The outer-join ListValueStore changelog stores the whole list blob,
always in the PLAIN
+ // element format, with the per-element headers parked in a reserved
record header. Restoring
+ // means re-inlining them. Legacy records written before the headers
format simply lack that
+ // header, which is the same as "every element has empty headers" — so
there is one path, not
+ // two. A tombstone (null value) is passed through.
+ if (record.value() == null) {
+ return record;
+ }
+
+ final byte[] convertedValue =
ListValueStoreUpgradeUtils.joinPlainListBlobWithElementHeaders(
+ record.value(),
+ ListValueStoreUpgradeUtils.elementHeaders(record.headers())
+ );
+
+ return new ConsumerRecord<>(
+ record.topic(),
+ record.partition(),
+ record.offset(),
+ record.timestamp(),
+ record.timestampType(),
+ record.serializedKeySize(),
+ convertedValue.length,
+ record.key(),
+ convertedValue,
+ record.headers(),
Review Comment:
I removed the `LIST_VALUE_HEADERS_HEADER_KEY`
--
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]
