aliehsaeedii commented on code in PR #22165:
URL: https://github.com/apache/kafka/pull/22165#discussion_r3558729010
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -437,19 +436,22 @@ public void evictWhile(final Supplier<Boolean> predicate,
@Override
public Maybe<ValueTimestampHeaders<V>> priorValueForBuffered(final K key) {
final Bytes serializedKey =
Bytes.wrap(keySerde.serializer().serialize(changelogTopic, context.headers(),
key));
- if (index.containsKey(serializedKey)) {
- final byte[] serializedValue =
internalPriorValueForBuffered(serializedKey);
+ final BufferKey bufferKey = index.get(serializedKey);
+ if (bufferKey != null) {
+ final BufferValue bufferValue = sortedMap.get(bufferKey);
+ final byte[] serializedValue = bufferValue.priorValue();
final V deserializedValue =
valueSerde.innerSerde().deserializer().deserialize(
changelogTopic,
- context.headers(),
+ bufferValue.context().headers(),
serializedValue
);
- // it's unfortunately not possible to know this, unless we
materialize the suppressed result, since our only
- // knowledge of the prior value is what the upstream processor
sends us as the "old value" when we first
- // buffer something.
- return Maybe.defined(ValueTimestampHeaders.make(deserializedValue,
RecordQueue.UNKNOWN, new RecordHeaders()));
+ return Maybe.defined(ValueTimestampHeaders.make(
+ deserializedValue,
+ bufferValue.context().timestamp(),
Review Comment:
Due to these explanations I reverted the changes of here. The former
[commit](https://github.com/apache/kafka/pull/22165/changes/87bee53184f012260b5bad0c293e653a3601c7e9#diff-b94df16c1e8d51db85b6539dce34da2e6cfda0a42584607d60c867e543f268e7)
returns `bufferValue.context().timestamp()` instead of `RecordQueue.UNKNOWN`
and `RecordHeaders(bufferValue.context().headers())` instead of empty headers.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedKeyValueBufferTest.java:
##########
@@ -306,10 +306,74 @@ public void shouldReturnPriorValueForBufferedKey(final
String testName, final Fu
context.setRecordContext(recordContext);
buffer.put(1L, new Record<>("A", new Change<>("new-value",
"old-value"), 0L), recordContext);
buffer.put(1L, new Record<>("B", new Change<>("new-value", null), 0L),
recordContext);
- assertThat(buffer.priorValueForBuffered("A"),
is(Maybe.defined(ValueTimestampHeaders.make("old-value", -1, new
RecordHeaders()))));
+ assertThat(buffer.priorValueForBuffered("A"),
is(Maybe.defined(ValueTimestampHeaders.make("old-value", 0L, new
RecordHeaders()))));
Review Comment:
These only cover a single put per key. The value-vs-metadata mismatch shows
up when the same key is put twice before eviction (prior value from put #1,
timestamp/headers from put #2) — worth a test for that case.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -437,19 +436,22 @@ public void evictWhile(final Supplier<Boolean> predicate,
@Override
public Maybe<ValueTimestampHeaders<V>> priorValueForBuffered(final K key) {
final Bytes serializedKey =
Bytes.wrap(keySerde.serializer().serialize(changelogTopic, context.headers(),
key));
Review Comment:
I'm wondering if `context.headers()` is the right thing for serializing the
key here and if not, how to replace it with the correct one?! Maybe just
sending empty headers?
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/TimeOrderedKeyValueBufferTest.java:
##########
@@ -306,10 +306,74 @@ public void shouldReturnPriorValueForBufferedKey(final
String testName, final Fu
context.setRecordContext(recordContext);
buffer.put(1L, new Record<>("A", new Change<>("new-value",
"old-value"), 0L), recordContext);
buffer.put(1L, new Record<>("B", new Change<>("new-value", null), 0L),
recordContext);
- assertThat(buffer.priorValueForBuffered("A"),
is(Maybe.defined(ValueTimestampHeaders.make("old-value", -1, new
RecordHeaders()))));
+ assertThat(buffer.priorValueForBuffered("A"),
is(Maybe.defined(ValueTimestampHeaders.make("old-value", 0L, new
RecordHeaders()))));
assertThat(buffer.priorValueForBuffered("B"), is(Maybe.defined(null)));
}
+ @ParameterizedTest
+ @MethodSource("parameters")
+ public void shouldPropagateHeadersThroughEviction(final String testName,
final Function<String, B> bufferSupplier) {
+ setup(testName, bufferSupplier);
+ final TimeOrderedKeyValueBuffer<String, String, Change<String>> buffer
= bufferSupplier.apply(testName);
+ final MockInternalProcessorContext<?, ?> context = makeContext();
+ buffer.init(context, buffer);
+
+ final RecordHeaders headers = new RecordHeaders(new Header[]{new
RecordHeader("h1", "v1".getBytes(UTF_8))});
Review Comment:
Every getBytes(...) call in this file specifies UTF_8 and UTF_8 is already
imported. Keeping getBytes(UTF_8) matches that convention and avoids relying on
the platform default charset.
--
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]
