vvcephei commented on a change in pull request #11213:
URL: https://github.com/apache/kafka/pull/11213#discussion_r703630483
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorSupplier.java
##########
@@ -198,7 +201,7 @@ private void emit(final
TimeOrderedKeyValueBuffer.Eviction<K, V> toEmit) {
final ProcessorRecordContext prevRecordContext =
internalProcessorContext.recordContext();
internalProcessorContext.setRecordContext(toEmit.recordContext());
try {
- internalProcessorContext.forward(toEmit.key(),
toEmit.value());
+ internalProcessorContext.forward(toEmit.record());
Review comment:
I think this might actually be a bug. IIRC, the Record forward call
overrides the context, so you might have to actually set all the fields in
Record from the context when you construct it in `toEmit.record()`.
Specifically, I think we might be dropping the headers here.
I might also be wrong, so you might want to double-check me first.
It looks like we were never testing whether we propagate headers through
Suppress. If it turns out that we never were, then there's also no problem. The
case to look out for is if we _were_, but now we no longer are.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
