mjsax commented on code in PR #22166:
URL: https://github.com/apache/kafka/pull/22166#discussion_r3592087480
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java:
##########
@@ -205,6 +218,87 @@ public void shouldHandleCollidingKeys() {
assertNumSizeAndTimestamp(buffer, 1, 7, 42);
}
+ @Test
+ public void shouldDeserializeWithPutTimeHeadersEvenAfterContextMutation() {
+ final HeaderCapturingSerde serde = new HeaderCapturingSerde();
+ createBuffer(Duration.ZERO, serde);
+ final RecordHeaders putHeaders = new RecordHeaders(new Header[]{
+ new RecordHeader("at-put",
"first".getBytes(StandardCharsets.UTF_8))
+ });
+ context.setRecordContext(new ProcessorRecordContext(0L, offset++, 0,
"testing", putHeaders));
Review Comment:
Should we set a different header on the context than on the record?
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java:
##########
@@ -205,6 +218,87 @@ public void shouldHandleCollidingKeys() {
assertNumSizeAndTimestamp(buffer, 1, 7, 42);
}
+ @Test
+ public void shouldDeserializeWithPutTimeHeadersEvenAfterContextMutation() {
+ final HeaderCapturingSerde serde = new HeaderCapturingSerde();
+ createBuffer(Duration.ZERO, serde);
+ final RecordHeaders putHeaders = new RecordHeaders(new Header[]{
+ new RecordHeader("at-put",
"first".getBytes(StandardCharsets.UTF_8))
+ });
+ context.setRecordContext(new ProcessorRecordContext(0L, offset++, 0,
"testing", putHeaders));
+ buffer.put(0L, new Record<>("k", "v", 0L, putHeaders),
context.recordContext());
+
+ // Simulate the processor moving on to another record with different
headers before eviction runs.
+ final RecordHeaders laterHeaders = new RecordHeaders(new Header[]{
+ new RecordHeader("at-evict",
"second".getBytes(StandardCharsets.UTF_8))
+ });
+ context.setRecordContext(new ProcessorRecordContext(0L, offset++, 0,
"testing", laterHeaders));
+
+ final List<TimeOrderedKeyValueBuffer.Eviction<String, String>> evicted
= new ArrayList<>();
+ buffer.evictWhile(() -> buffer.numRecords() > 0, evicted::add);
+
+ assertThat(evicted.size(), is(1));
+ // The key/value deserializers must see the headers captured at put
time, not the mutated context headers.
+ assertThat(serde.capturedHeaders, hasItem(putHeaders));
+ assertThat(serde.capturedHeaders, everyItem(is(putHeaders)));
+ assertThat(evicted.get(0).recordContext().headers(), is(putHeaders));
+ }
+
+ @Test
+ public void
shouldNotBeAffectedByProcessorContextHeaderMutationBetweenPutAndEvict() {
+ final HeaderCapturingSerde serde = new HeaderCapturingSerde();
+ createBuffer(Duration.ofMillis(1), serde);
+ final RecordHeaders putHeaders = new RecordHeaders(new Header[]{
+ new RecordHeader("at-put",
"first".getBytes(StandardCharsets.UTF_8))
+ });
+ context.setRecordContext(new ProcessorRecordContext(0L, offset++, 0,
"testing", putHeaders));
+ buffer.put(0L, new Record<>("k", "v", 0L, putHeaders),
context.recordContext());
+
+ // Simulate the processor moving on to handle a different record with
different headers
+ // before the grace period expires and eviction runs.
+ final RecordHeaders laterHeaders = new RecordHeaders(new Header[]{
+ new RecordHeader("at-evict",
"second".getBytes(StandardCharsets.UTF_8))
+ });
+ context.setRecordContext(new ProcessorRecordContext(10L, offset++, 0,
"testing", laterHeaders));
Review Comment:
Same here?
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java:
##########
@@ -205,6 +218,87 @@ public void shouldHandleCollidingKeys() {
assertNumSizeAndTimestamp(buffer, 1, 7, 42);
}
+ @Test
+ public void shouldDeserializeWithPutTimeHeadersEvenAfterContextMutation() {
+ final HeaderCapturingSerde serde = new HeaderCapturingSerde();
+ createBuffer(Duration.ZERO, serde);
+ final RecordHeaders putHeaders = new RecordHeaders(new Header[]{
+ new RecordHeader("at-put",
"first".getBytes(StandardCharsets.UTF_8))
+ });
+ context.setRecordContext(new ProcessorRecordContext(0L, offset++, 0,
"testing", putHeaders));
+ buffer.put(0L, new Record<>("k", "v", 0L, putHeaders),
context.recordContext());
+
+ // Simulate the processor moving on to another record with different
headers before eviction runs.
+ final RecordHeaders laterHeaders = new RecordHeaders(new Header[]{
+ new RecordHeader("at-evict",
"second".getBytes(StandardCharsets.UTF_8))
+ });
+ context.setRecordContext(new ProcessorRecordContext(0L, offset++, 0,
"testing", laterHeaders));
+
+ final List<TimeOrderedKeyValueBuffer.Eviction<String, String>> evicted
= new ArrayList<>();
+ buffer.evictWhile(() -> buffer.numRecords() > 0, evicted::add);
+
+ assertThat(evicted.size(), is(1));
+ // The key/value deserializers must see the headers captured at put
time, not the mutated context headers.
+ assertThat(serde.capturedHeaders, hasItem(putHeaders));
+ assertThat(serde.capturedHeaders, everyItem(is(putHeaders)));
+ assertThat(evicted.get(0).recordContext().headers(), is(putHeaders));
+ }
+
+ @Test
+ public void
shouldNotBeAffectedByProcessorContextHeaderMutationBetweenPutAndEvict() {
+ final HeaderCapturingSerde serde = new HeaderCapturingSerde();
+ createBuffer(Duration.ofMillis(1), serde);
+ final RecordHeaders putHeaders = new RecordHeaders(new Header[]{
+ new RecordHeader("at-put",
"first".getBytes(StandardCharsets.UTF_8))
+ });
+ context.setRecordContext(new ProcessorRecordContext(0L, offset++, 0,
"testing", putHeaders));
Review Comment:
Same question
--
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]
