Re: [PR] KAFKA-20413: Make Streams suppress headers-aware [kafka]

["aliehsaeedii (via GitHub)" <gi...@apache.org>]

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: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org