aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3638418494
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ private byte[] internalPriorValueForBuffered(final Bytes
key) {
}
}
+ // The legacy V0/V1 restore paths feed the currently-buffered prior value
back in as plain value
+ // bytes, so unwrap it when the in-memory encoding is the headers-aware
format.
+ private byte[] plainPriorValueForBuffered(final Bytes key) {
+ final byte[] priorValue = internalPriorValueForBuffered(key);
+ return storeHeaders ? unwrapHeadersFormatToPlainValue(priorValue) :
priorValue;
+ }
+
+ // Normalizes a restored buffer value so its value parts match the
encoding currently used
+ // in-memory (ValueTimestampHeaders blobs when header stores are enabled,
plain values otherwise),
+ // regardless of the changelog version it was read from.
+ private BufferValue normalizeEncoding(final BufferValue value, final
boolean restoredWithHeaders) {
+ if (storeHeaders == restoredWithHeaders) {
+ return value;
+ }
+ if (storeHeaders) {
+ // Plain -> ValueTimestampHeaders. The original headers/timestamp
of legacy records are
+ // unknown, so we use empty headers and the record-context
timestamp.
+ final long timestamp = value.context().timestamp();
+ return new BufferValue(
+ wrapPlainValueAsHeadersFormat(value.priorValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.oldValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.newValue(), timestamp),
+ value.context()
+ );
+ }
+ // ValueTimestampHeaders -> plain would drop the per-value
headers/timestamps carried by a V4
+ // changelog. This only arises when a store that previously used a
headers store format is
+ // restored without one, i.e. a store-format downgrade. Downgrades are
not supported, so we
+ // fail fast here rather than silently discarding the stored header
data.
+ throw new StreamsException(
Review Comment:
Ok, of course we can, But why???? we do not support downgrade fo rthe whole
thing. Why should we do that in this small corner? it is confusing as is not
consistent with the rest of the code and the KIPs. WDYT?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +513,165 @@ private byte[] internalPriorValueForBuffered(final Bytes
key) {
}
}
+ // The legacy V0/V1 restore paths feed the currently-buffered prior value
back in as plain value
+ // bytes, so unwrap it when the in-memory encoding is the headers-aware
format.
+ private byte[] plainPriorValueForBuffered(final Bytes key) {
+ final byte[] priorValue = internalPriorValueForBuffered(key);
+ return storeHeaders ? unwrapHeadersFormatToPlainValue(priorValue) :
priorValue;
+ }
+
+ // Normalizes a restored buffer value so its value parts match the
encoding currently used
+ // in-memory (ValueTimestampHeaders blobs when header stores are enabled,
plain values otherwise),
+ // regardless of the changelog version it was read from.
+ private BufferValue normalizeEncoding(final BufferValue value, final
boolean restoredWithHeaders) {
+ if (storeHeaders == restoredWithHeaders) {
+ return value;
+ }
+ if (storeHeaders) {
+ // Plain -> ValueTimestampHeaders. The original headers/timestamp
of legacy records are
+ // unknown, so we use empty headers and the record-context
timestamp.
+ final long timestamp = value.context().timestamp();
+ return new BufferValue(
+ wrapPlainValueAsHeadersFormat(value.priorValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.oldValue(), timestamp),
+ wrapPlainValueAsHeadersFormat(value.newValue(), timestamp),
+ value.context()
+ );
+ }
+ // ValueTimestampHeaders -> plain would drop the per-value
headers/timestamps carried by a V4
+ // changelog. This only arises when a store that previously used a
headers store format is
+ // restored without one, i.e. a store-format downgrade. Downgrades are
not supported, so we
+ // fail fast here rather than silently discarding the stored header
data.
+ throw new StreamsException(
Review Comment:
Ok, of course we can, But why? we do not support downgrade fo rthe whole
thing. Why should we do that in this small corner? it is confusing as is not
consistent with the rest of the code and the KIPs. WDYT?
--
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]
