aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3664770759
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -203,6 +222,10 @@ public void init(final StateStoreContext
stateStoreContext, final StateStore roo
taskId = context.taskId().toString();
streamsMetrics = context.metrics();
+ final Object dslStoreFormat =
stateStoreContext.appConfigs().get(StreamsConfig.DSL_STORE_FORMAT_CONFIG);
Review Comment:
Good point. We generally must update the docs after this PR:
https://issues.apache.org/jira/browse/KAFKA-20849
>Should we file a Jira ticket about "allow enabling header-support on
suppress() similar to Materialized for other operators"?
Mmm. Does it need a KIP? does it override dsl.store.format? I don't fully
understand this.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -399,14 +448,43 @@ public void evictWhile(final Supplier<Boolean> predicate,
next.getKey().time() + "]"
);
}
- final K key =
keySerde.deserializer().deserialize(changelogTopic, context.headers(),
next.getKey().key().get());
final BufferValue bufferValue = next.getValue();
- final Change<V> value = valueSerde.deserializeParts(
- changelogTopic,
- context.headers(),
- new Change<>(bufferValue.newValue(),
bufferValue.oldValue())
- );
- callback.accept(new Eviction<K, Change<V>>(key, value,
bufferValue.context()));
+ final ProcessorRecordContext bufferedContext =
bufferValue.context();
+
+ final K key;
+ final Change<V> value;
+ final ProcessorRecordContext evictionContext;
+ if (storeHeaders) {
+ // Each value part was stored as a ValueTimestampHeaders
blob carrying the headers of
+ // the record it originated from, so take them from the
value rather than from the
+ // buffered context: the old value came from an earlier
record than the new one, and
+ // deserializing it with the new record's headers would
hand the serde the wrong ones.
+ // (ValueTimestampHeadersDeserializer feeds each part's
own headers to the inner
+ // value deserializer.)
+ final ValueTimestampHeaders<V> newPart =
deserializeValuePart(bufferValue.newValue());
Review Comment:
"Part" means prior/old/new. The three value parts of a BufferValue — but the
method itself knows nothing about that. it's just raw VTH bytes →
ValueTimestampHeaders. So the name describes the caller, not the callee. I
rename it to `deserializeValueTimestampHeaders`
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -465,28 +569,135 @@ 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 ? Utils.rawPlainValue(priorValue) : priorValue;
+ }
+
+ // The changelog value format is plain for every version, so a restored
buffer value always has
+ // plain value parts. When header stores are enabled the in-memory
encoding is
+ // ValueTimestampHeaders, so re-attach each part's headers and timestamp
from the changelog
+ // record's own Kafka headers. Records written before this feature -- or
by a run configured
+ // without header stores -- carry no such headers, in which case the
originals are genuinely
+ // unknown and we fall back to empty headers with the record-context
timestamp.
+ private BufferValue toInMemoryEncoding(final BufferValue value, final
Headers recordHeaders) {
+ if (!storeHeaders) {
+ return value;
+ }
+ final long fallbackTimestamp = value.context().timestamp();
+ return new BufferValue(
+ valuePartWithHeaders(value.priorValue(), recordHeaders,
PRIOR_VALUE_HEADERS_KEY, fallbackTimestamp),
+ valuePartWithHeaders(value.oldValue(), recordHeaders,
OLD_VALUE_HEADERS_KEY, fallbackTimestamp),
+ valuePartWithHeaders(value.newValue(), recordHeaders,
NEW_VALUE_HEADERS_KEY, fallbackTimestamp),
+ value.context()
+ );
+ }
+
+ private static byte[] valuePartWithHeaders(final byte[] rawPlainValue,
+ final Headers recordHeaders,
+ final String headerKey,
+ final long fallbackTimestamp) {
+ if (rawPlainValue == null) {
+ return null;
+ }
+ final Header prefix = recordHeaders.lastHeader(headerKey);
Review Comment:
It is a Header object whose value is the [headersSize][headers][timestamp]
byte prefix. I renamed it to
`headersAndTimestamp`
--
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]
